timxx/Yangmao_Script
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

更新 Cash_Based/JiaShan.py

Browse Source
This commit is contained in:
admin
2026-06-06 03:52:27 +08:00
parent 95c81747b6
commit 6b5e264b07
1 changed files with 198 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

264
Cash_Based/JiaShan.py
View File

@@ -1,10 +1,9 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
in嘉善 JiaShan 明文 Python 版
# cron: 31 9,19 * * *
# const $ = new Env('in嘉善阅读');
cron: 31 9,19 * * *
"""
in嘉善 JiaShan 明文 Python 版
环境变量：
JiaShan="账号1&密码1\n账号2&密码2"
@@ -34,9 +33,11 @@ import random
import re
import time
import uuid
import hashlib
from dataclasses import dataclass, field
from datetime import datetime, timezone, timedelta
from typing import Any, Dict, List, Optional
from urllib.parse import unquote
from urllib.parse import unquote, quote_plus
try:
import requests
@@ -46,15 +47,20 @@ except ImportError:
NAME = "in嘉善"
BASE = "https://api.app.injs.jsxww.cn"
YAPI_BASE = "https://yapi.y-h5.iyunxh.com/api"
OAPI_BASE = "https://oapi.injs.jsxww.cn"
DEBUG = os.getenv("JIASHAN_DEBUG", "false").lower() in {"1", "true", "yes"}
TIMEOUT = int(os.getenv("JIASHAN_TIMEOUT", "20") or "20")
TASK_DELAY_MIN = float(os.getenv("JIASHAN_TASK_DELAY_MIN", "5") or "5")
TASK_DELAY_MAX = float(os.getenv("JIASHAN_TASK_DELAY_MAX", "10") or "10")
ACCOUNT_DELAY_MIN = float(os.getenv("JIASHAN_ACCOUNT_DELAY_MIN", "8") or "8")
ACCOUNT_DELAY_MAX = float(os.getenv("JIASHAN_ACCOUNT_DELAY_MAX", "15") or "15")
CN_TZ = timezone(timedelta(hours=8))
# 原 JS 追踪到的首页动态组件文章列表入口
# 原 JS 追踪到的阅读活动入口。这个接口返回的不是普通新闻列表，而是活动页 q/id。
ARTICLE_LIST_PATH = "/app/layout/dynamic/component/data?layoutId=7853114638635438077&layoutDatasourceId=7853114638635438096&pageNo=1&pageSize=20"
READ_ACTIVITY_ID = os.getenv("JIASHAN_READ_ACTIVITY_ID", "11106624")
TENANT_CODE = os.getenv("JIASHAN_TENANT_CODE", "in_jiashan")
MOBILE_UA = (
"Mozilla/5.0 (Linux; Android 11; 21091116AC Build/RP1A.200720.011; wv) "
@@ -63,6 +69,15 @@ MOBILE_UA = (
)
def now_cn() -> datetime:
return datetime.now(CN_TZ)
def cn_month_day() -> str:
now = now_cn()
return f"{now.month}月{now.day}日"
def mask(s: Any, keep: int = 4) -> str:
s = "" if s is None else str(s)
if len(s) <= keep * 2:
@@ -107,8 +122,53 @@ def js_wait(label: str = "") -> None:
def extract_first_q(obj: Any) -> str:
m = re.search(r'(?<=q=)[^&",]+', json.dumps(obj, ensure_ascii=False))
return unquote(m.group(0)) if m else ""
text = json.dumps(obj, ensure_ascii=False)
# 原 JS 需要的是活动链接里的 q 参数；不能随便取 layoutDatasourceId/componentId，
# 否则 activity-api 会报“链接解密失败”。
for pat in [r'(?<=q=)[^&",\\]+', r'(?<=[?&]q%3D)[^%&",\\]+']:
m = re.search(pat, text)
if m:
return unquote(m.group(0))
# 只有明确的 q 字段才当活动 q 使用。
def walk(obj2: Any):
if isinstance(obj2, dict):
if obj2.get("q"):
yield obj2.get("q")
for key in ("url", "jumpUrl", "linkUrl", "h5Url", "targetUrl", "externalUrl", "activityUrl"):
val = obj2.get(key)
if isinstance(val, str):
mm = re.search(r'(?<=q=)[^&",\\]+', val)
if mm:
yield unquote(mm.group(0))
for v in obj2.values():
yield from walk(v)
elif isinstance(obj2, list):
for v in obj2:
yield from walk(v)
for val in walk(obj):
val = str(val).strip()
if val:
return unquote(val)
return ""
def random_string(length: int = 32, prefix_u: bool = True, radix: Optional[int] = None) -> str:
chars = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
if radix is None:
radix = len(chars)
out = "".join(chars[int(random.random() * radix)] for _ in range(length))
return ("u" + out[1:]) if prefix_u and out else out
def urlencode_form(obj: Dict[str, Any]) -> str:
return "&".join(f"{k}={quote_plus(str(v))}" for k, v in obj.items())
def yapi_signature() -> str:
ts = int(time.time() * 1000)
nonce = random_string(32, prefix_u=False)
raw = f"jsxww{nonce}{ts}3a82b6ac78145c2a6c4ff1f7d3dced1b"
return f"jsxww;{nonce};{ts};{hashlib.md5(raw.encode()).hexdigest()}"
def walk_values(obj: Any):
@@ -135,6 +195,8 @@ class ClientState:
user_id: str = ""
read_q: str = ""
read_token: str = ""
yapi_user_id: str = "0"
api_dt: str = ""
lottery_q: str = ""
lottery_token: str = ""
third_id: str = ""
@@ -196,7 +258,7 @@ class JiaShanRunner:
url = path if path.startswith("http") else BASE + path
return self.request_json("POST", url, state=state, json_body=payload, desc=path)
def login(self, account: Account, state: ClientState) -> bool:
def login(self, account: Account, state: ClientState) -> Optional[Dict[str, Any]]:
print("登录")
ret = self.request_json(
"POST",
@@ -214,81 +276,150 @@ class JiaShanRunner:
user_id = str(data.get("userId") or data.get("uid") or data.get("id") or user.get("id") or user.get("userId") or user.get("uid") or "")
if not token:
print(f"登录失败：{ret.get('message') or ret.get('msg') or short_json(ret)}")
return False
return None
state.token = str(token)
state.user_id = user_id
print(f"登录成功：token={mask(state.token)}" + (f" user_id={mask(user_id)}" if user_id else ""))
return data
def yapi_headers(self, state: ClientState, authed: bool = True) -> Dict[str, str]:
h = {
"Connection": "Keep-Alive",
"Access-T-Id-In": "49",
"Access-Wxclient-Type": "wx_app",
"User-Agent": "injs;android:11;version:1.1.12;clientid:" + state.client_id + ";" + MOBILE_UA,
"Access-Api-Unique-Token": "1",
"Access-Api-Dt": state.api_dt or str(int(time.time() * 1000)),
"Access-T-Id": "49",
"Accept": "*/*",
"Origin": "https://jsxww.y-h5.iyunxh.com",
"X-Requested-With": "info.ltit.www.cloudjiashan",
"Referer": "https://jsxww.y-h5.iyunxh.com/",
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7",
}
if authed:
h["Access-User-Id"] = state.yapi_user_id or "0"
h["Access-Api-Signature"] = yapi_signature()
h["Access-Token"] = state.read_token
return h
def yapi_get(self, state: ClientState, path: str, *, authed: bool = True, wait: bool = True) -> Dict[str, Any]:
return self.request_json("GET", YAPI_BASE + path, headers=self.yapi_headers(state, authed), desc=path, wait=wait)
def yapi_post(self, state: ClientState, path: str, payload: Dict[str, Any], *, wait: bool = True) -> Dict[str, Any]:
headers = self.yapi_headers(state, True)
headers["Content-Type"] = "application/json"
return self.request_json("POST", YAPI_BASE + path, headers=headers, json_body=payload, desc=path, wait=wait)
def oapi_get(self, state: ClientState, path: str, *, wait: bool = True) -> Dict[str, Any]:
headers = {
"Connection": "Keep-Alive",
"ClientId": state.client_id,
"Authorization": state.token,
"User-Agent": "injs;android:11;version:1.1.12;clientid:" + state.client_id,
"Accept-Encoding": "gzip",
}
return self.request_json("GET", OAPI_BASE + path, headers=headers, desc=path, wait=wait)
def app_user_init(self, account: Account, state: ClientState, app_user_token: str, login_data: Dict[str, Any]) -> bool:
userinfo = login_data.get("userinfo") or login_data.get("userInfo") or login_data.get("user") or {}
user_id = userinfo.get("id") or state.user_id
nickname = userinfo.get("nickname") or userinfo.get("name") or userinfo.get("username") or f"用户{account.phone[-6:]}"
avatar = userinfo.get("avatarUrl") or userinfo.get("avatar_url") or ""
payload = {
"app_user_token": app_user_token,
"appid": "jsxww",
"noncestr": random_string(6, prefix_u=False),
"phone": account.phone,
"portrait_url": ("https://oss.injs.jsxww.cn" + avatar) if avatar and str(avatar).startswith("/") else str(avatar),
"timestamp": str(int(time.time())),
"user_id": user_id,
"user_name": nickname,
"wx_openid": "",
"wx_unionid": "",
}
payload["signature"] = hashlib.md5((urlencode_form(payload) + "&appkey=0c3eafb13e9f1ac110a432798b021862").encode()).hexdigest()
ret = self.yapi_post(state, "/aosbase/_auth_appuserinit", payload)
data = ret.get("data") or {}
state.read_token = data.get("access_token") or data.get("token") or ""
state.yapi_user_id = str((data.get("data") or {}).get("user_id") or data.get("user_id") or "0")
if not state.read_token:
print(f"阅读登录失败：{short_json(ret)}")
return False
print(f"阅读token：{state.read_token}")
return True
def pick_article_items(self, obj: Dict[str, Any]) -> List[Dict[str, Any]]:
candidates = []
data = obj.get("data")
for val in [data, *list(walk_values(data))]:
if isinstance(val, list):
dicts = [x for x in val if isinstance(x, dict)]
if dicts and any(("id" in x or "articleId" in x or "contentId" in x) for x in dicts):
candidates.append(dicts)
return candidates[0] if candidates else []
def article_id_of(self, article: Dict[str, Any]) -> str:
return str(article.get("id") or article.get("articleId") or article.get("contentId") or article.get("resourceId") or "")
def do_read_tasks_and_lottery(self, account: Account, state: ClientState) -> None:
def do_read_tasks_and_lottery(self, account: Account, state: ClientState, login_data: Dict[str, Any]) -> None:
print("————————————")
print("阅读抽奖")
print("获取id")
article_list = self.api_get(state, ARTICLE_LIST_PATH)
state.read_q = extract_first_q(article_list)
articles = self.pick_article_items(article_list)
if state.read_q:
print(f"获取id完成：{mask(state.read_q)}")
elif articles:
print(f"获取文章列表完成：{len(articles)}篇")
else:
print("获取id失败")
if not state.read_q:
# 历史原 JS 日志中此活动 q 为 11106624；仅在页面未暴露 q 时兜底使用。
state.read_q = READ_ACTIVITY_ID
print(state.read_q)
print("获取apiDt")
auth_dt = self.yapi_get(state, "/aosbase/_auth_dt", authed=False)
state.api_dt = str(auth_dt.get("data") or "")[32:68]
print(state.api_dt)
if not state.api_dt:
print(f"获取apiDt失败：{short_json(auth_dt)}")
return
# in嘉善原脚本主要是阅读抽奖。若活动接口与天目云一致，则按同类接口尝试；失败不影响阅读。
if state.read_q:
print("获取阅读token")
auth = self.request_json(
"POST",
"https://act.tmlyun.com/activity-api/task/h5/auth/userLogin",
state=state,
json_body={"q": state.read_q, "accountId": state.user_id, "sessionId": state.token, "tenantCode": "in_jiashan"},
headers={"Authorization": state.read_token, "X-Requested-With": "com.jsxww.injiashan"},
desc="获取阅读token",
)
state.read_token = (auth.get("data") or {}).get("token") or ""
if state.read_token:
print(f"获取阅读token完成：{mask(state.read_token)}")
# 阅读文章：不同版本接口字段可能不同，这里按常见阅读上报接口做兼容尝试。
for i, article in enumerate(articles, 1):
aid = self.article_id_of(article)
if not aid:
continue
ret = self.api_get(state, f"/app/article/read_time?channel_article_id={aid}&is_end=1&read_time=1617")
print(f"阅读{i}/{len(articles)}：{ret.get('message') or ret.get('msg') or short_json(ret, 120)}")
if not state.read_token:
print("阅读登录")
access_ret = self.yapi_get(state, "/admin/_service_custom_jsxww_getaccesstoken?access_t_id=1&access_t_id_in=1", authed=False)
access_token = (access_ret.get("data") or "")
open_ret = self.oapi_get(state, f"/auth/openid?access_token={access_token}&app_token={state.token}")
open_data = open_ret.get("data") or {}
app_user_token = f"{open_data.get('openid')}.{open_data.get('ticket')}" if open_data.get("openid") and open_data.get("ticket") else ""
if not app_user_token or not self.app_user_init(account, state, app_user_token, login_data):
print("未获取到活动token，跳过抽奖")
return
lottery_info = self.request_json(
"GET",
"https://act.tmlyun.com/activity-api/task/h5/activity/getLotteryInfo",
state=state,
headers={"Authorization": state.read_token, "X-Requested-With": "com.jsxww.injiashan"},
desc="获取抽奖次数",
)
today_md = cn_month_day()
pass_list_ret = self.yapi_get(state, f"/aoslearnfoot/_optionp_list?activity_id={state.read_q}")
pass_list = pass_list_ret.get("data") or []
if isinstance(pass_list, dict):
pass_list = pass_list.get("list") or pass_list.get("records") or []
for item in pass_list if isinstance(pass_list, list) else []:
pass_id = item.get("id")
title = item.get("title") or item.get("name") or f"{today_md}阅读"
print(title)
detail = self.yapi_get(state, f"/aoslearnfoot/optionp_detail?id={pass_id}")
data = detail.get("data") or {}
task_num = int(data.get("task_num") or item.get("task_num") or 0)
done_num = int(data.get("user_done_num") or item.get("user_done_num") or 0)
if task_num and task_num == done_num:
print("已完成")
continue
print(f"进度：{done_num}/{task_num}")
# 需要滑块时，原 JS 会请求 ddddocr。Python 版先不伪造滑块，避免错误提交。
if task_num != done_num:
print("未完成任务需要滑块验证，已跳过阅读提交")
if not pass_list:
print(f"{today_md}阅读")
print("未获取到阅读任务")
ac_detail = self.yapi_get(state, f"/aoslearnfoot/_ac_detail?id={state.read_q}")
lottery_id = ""
try:
lottery_count = int((lottery_info.get("data") or {}).get("lotteryCount") or 0)
other_set = json.loads((ac_detail.get("data") or {}).get("other_set") or "{}")
lottery_id = str(((other_set.get("lottery") or {}).get("id")) or "")
except Exception:
lottery_id = ""
lottery_times = self.yapi_get(state, f"/aoslottery/ac_lottery_times?id={lottery_id}") if lottery_id else {}
try:
lottery_count = int((lottery_times.get("data") or {}).get("all_remain") or 0)
except Exception:
lottery_count = 0
print(f"拥有{lottery_count}次抽奖")
if lottery_count <= 0:
return
print("当前账号有抽奖次数，但抽奖接口需要滑块验证，Python版已跳过自动抽奖")
return
print("获取抽奖token")
activity_info = self.request_json(
@@ -303,7 +434,7 @@ class JiaShanRunner:
"POST",
"https://act.tmlyun.com/activity-api/lottery/api/auth/userLogin",
state=state,
json_body={"q": state.lottery_q, "accountId": state.user_id, "sessionId": state.token, "tenantCode": "in_jiashan"},
json_body={"q": state.lottery_q, "accountId": state.user_id, "sessionId": state.token, "tenantCode": TENANT_CODE},
headers={"Authorization": state.lottery_token, "X-Requested-With": "com.jsxww.injiashan"},
desc="获取抽奖token",
)
@@ -332,9 +463,10 @@ class JiaShanRunner:
print("随机生成clientId")
print(state.client_id)
print(f"用户：{account.phone}开始任务")
if not self.login(account, state):
login_data = self.login(account, state)
if not login_data:
return state.summary_lines
self.do_read_tasks_and_lottery(account, state)
self.do_read_tasks_and_lottery(account, state, login_data)
return state.summary_lines
@@ -388,4 +520,4 @@ def main() -> int:
if __name__ == "__main__":
raise SystemExit(main())
raise SystemExit(main())