783 lines
34 KiB
Python
783 lines
34 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
studycourse-analysis — 角色上课情况分析工具
|
||
用法: python3 studycourse_analysis.py <role_id> [--format html] [--output <path>]
|
||
"""
|
||
|
||
import os, sys, argparse, json
|
||
import psycopg2, psycopg2.extras, pymysql
|
||
from datetime import datetime, timedelta
|
||
from collections import defaultdict, OrderedDict
|
||
|
||
# ── 配置 ──
|
||
PG_CONFIG = {
|
||
"host": os.environ.get("PG_DB_HOST", "bj-postgres-16pob4sg.sql.tencentcdb.com"),
|
||
"port": int(os.environ.get("PG_DB_PORT", "28591")),
|
||
"user": os.environ.get("PG_DB_USER", "ai_member"),
|
||
"password": os.environ.get("PG_DB_PASSWORD", ""),
|
||
"dbname": os.environ.get("PG_DB_DATABASE", "vala"),
|
||
}
|
||
MYSQL_HOST = os.environ.get("MYSQL_HOST_online", "bj-cdb-dh2fkqa0.sql.tencentcdb.com")
|
||
MYSQL_PORT = int(os.environ.get("MYSQL_PORT_online", "27751"))
|
||
MYSQL_USER = os.environ.get("MYSQL_USERNAME_online", "read_only")
|
||
MYSQL_PASS = os.environ.get("MYSQL_PASSWORD_online", "")
|
||
|
||
WEEKDAY_NAMES = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
|
||
|
||
|
||
# ── 数据库连接 ──
|
||
def pg_conn():
|
||
return psycopg2.connect(host=PG_CONFIG["host"], port=PG_CONFIG["port"],
|
||
user=PG_CONFIG["user"], password=PG_CONFIG["password"], dbname=PG_CONFIG["dbname"])
|
||
|
||
def my_conn(db="vala_user"):
|
||
return pymysql.connect(host=MYSQL_HOST, port=MYSQL_PORT, user=MYSQL_USER,
|
||
password=MYSQL_PASS, db=db, charset="utf8mb4")
|
||
|
||
|
||
# ── 章节全局单元映射（复用 studytime-analysis 逻辑） ──
|
||
_chapter_map = None
|
||
|
||
def fetch_chapter_info_map():
|
||
global _chapter_map
|
||
if _chapter_map is not None:
|
||
return _chapter_map
|
||
conn = my_conn("vala")
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute("""SELECT gc.id, IFNULL(sp.level,''), IFNULL(sp.season_of_quarter,-1),
|
||
gc.`index`, gc.lesson_type
|
||
FROM vala_game_chapter gc
|
||
LEFT JOIN vala_game_season_package sp ON gc.season_package_id=sp.id
|
||
ORDER BY sp.level, sp.season_of_quarter, gc.id""")
|
||
rows = cur.fetchall()
|
||
finally:
|
||
conn.close()
|
||
from collections import OrderedDict as OD
|
||
seasons = OD()
|
||
for ch_id, lv, sq, li, lt in rows:
|
||
sq = int(sq) if sq is not None else -1
|
||
lt = int(lt) if lt is not None else 1
|
||
key = (lv or "", sq)
|
||
seasons.setdefault(key, []).append((int(ch_id), lv or "", int(li or 0), lt))
|
||
def base(s):
|
||
return 0 if s <= 0 else 1 + 12 * (s - 1)
|
||
_chapter_map = {}
|
||
for (lv_key, sq), ch_list in seasons.items():
|
||
regular = [(cid, lv, li) for cid, lv, li, lt in ch_list if lt == 1]
|
||
uid_map = {}
|
||
if sq <= 0:
|
||
for cid, lv, li in regular:
|
||
uid_map[cid] = 0
|
||
else:
|
||
for pos, (cid, lv, li) in enumerate(regular):
|
||
uid_map[cid] = pos // 5
|
||
for cid, lv, li, lt in ch_list:
|
||
u = uid_map.get(cid, -1)
|
||
_chapter_map[cid] = dict(level=lv,
|
||
unit_index=base(sq) + u if u >= 0 else -1,
|
||
lesson_index=li, lesson_type=lt)
|
||
return _chapter_map
|
||
|
||
|
||
# ── 数据查询 ──
|
||
|
||
def fetch_role_info(role_id):
|
||
conn = my_conn("vala_user")
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute("""SELECT c.id, c.account_id, c.nickname, c.gender, c.birthday,
|
||
c.created_at, c.purchase_season_package
|
||
FROM vala_app_character c WHERE c.id=%s""", (role_id,))
|
||
row = cur.fetchone()
|
||
finally:
|
||
conn.close()
|
||
if not row:
|
||
return None
|
||
rid, aid, nn, gd, bd, crt, psp = row
|
||
gs = {0: "女", 1: "男"}.get(gd, str(gd) if gd is not None else "")
|
||
age = ""
|
||
if bd:
|
||
try: age = datetime.now().year - int(str(bd).split("-")[0])
|
||
except: pass
|
||
return dict(role_id=rid, account_id=aid, nickname=nn or "", gender=gs, age=age,
|
||
reg_time=crt.strftime("%Y-%m-%d %H:%M") if crt else "-",
|
||
purchase_packages=str(psp) if psp else "")
|
||
|
||
|
||
def fetch_account_info(account_id):
|
||
conn = my_conn("vala_user")
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute("""SELECT id, tel, download_channel, key_from, created_at, pay_status
|
||
FROM vala_app_account WHERE id=%s""", (account_id,))
|
||
row = cur.fetchone()
|
||
finally:
|
||
conn.close()
|
||
if not row:
|
||
return {}
|
||
aid, tel, ch, kf, crt, ps = row
|
||
pt = ""
|
||
if tel:
|
||
d = ''.join(c for c in str(tel) if c.isdigit())
|
||
pt = d[-4:] if len(d) >= 4 else d
|
||
return dict(account_id=aid, phone_tail=pt, download_channel=ch or "-",
|
||
key_from=kf or "-", account_created=crt.strftime("%Y-%m-%d %H:%M") if crt else "-",
|
||
pay_status=ps)
|
||
|
||
|
||
def fetch_course_detail(role_id):
|
||
conn = pg_conn()
|
||
try:
|
||
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
||
cur.execute("SELECT * FROM user_course_detail WHERE user_id=%s ORDER BY id LIMIT 1", (role_id,))
|
||
return cur.fetchone()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def fetch_device_info(account_id):
|
||
conn = pg_conn()
|
||
try:
|
||
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
||
cur.execute("SELECT * FROM user_login_app_info WHERE account_id=%s ORDER BY created_at DESC LIMIT 1",
|
||
(account_id,))
|
||
return cur.fetchone()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def fetch_play_records(role_id):
|
||
parts, params = [], {}
|
||
for i in range(8):
|
||
pn = f"r{i}"
|
||
params[pn] = role_id
|
||
parts.append(f"""SELECT user_id, chapter_id, chapter_unique_id, level, created_at, updated_at
|
||
FROM user_chapter_play_record_{i} WHERE user_id=%({pn})s AND play_status=1""")
|
||
sql = f"SELECT * FROM ({' UNION ALL '.join(parts)}) t ORDER BY updated_at ASC"
|
||
conn = pg_conn()
|
||
try:
|
||
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
||
cur.execute(sql, params)
|
||
return cur.fetchall()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def fetch_component_records(role_id):
|
||
"""获取中互动记录"""
|
||
parts, params = [], {}
|
||
for i in range(8):
|
||
pn = f"r{i}"
|
||
params[pn] = role_id
|
||
parts.append(f"""SELECT user_id, chapter_id, section_id, component_unique_code,
|
||
play_result, skill_points, c_type, c_id, created_at, updated_at, pass_time,
|
||
read_word_count, speak_count, listen_sentence_count, write_word_count
|
||
FROM user_component_play_record_{i} WHERE user_id=%({pn})s""")
|
||
sql = f"SELECT * FROM ({' UNION ALL '.join(parts)}) t ORDER BY updated_at ASC"
|
||
conn = pg_conn()
|
||
try:
|
||
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
||
cur.execute(sql, params)
|
||
return cur.fetchall()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def fetch_settlement_data(role_id):
|
||
"""获取结算数据（含巩固题信息）"""
|
||
parts, params = [], {}
|
||
for i in range(8):
|
||
pn = f"r{i}"
|
||
params[pn] = role_id
|
||
parts.append(f"""SELECT user_id, chapter_id, unique_id, settlement_data, created_at, updated_at, level
|
||
FROM user_chapter_settlement_data_{i} WHERE user_id=%({pn})s""")
|
||
sql = f"SELECT * FROM ({' UNION ALL '.join(parts)}) t ORDER BY updated_at ASC"
|
||
conn = pg_conn()
|
||
try:
|
||
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
||
cur.execute(sql, params)
|
||
rows = cur.fetchall()
|
||
for r in rows:
|
||
if r.get("settlement_data"):
|
||
try:
|
||
r["sd_parsed"] = json.loads(r["settlement_data"])
|
||
except:
|
||
r["sd_parsed"] = {}
|
||
return rows
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
# ── 分析函数 ──
|
||
|
||
def analyze_duration(records):
|
||
"""分析完课耗时"""
|
||
durations = []
|
||
anomalies_short = []
|
||
anomalies_long = []
|
||
for r in records:
|
||
sd = r.get("created_at")
|
||
ed = r.get("updated_at")
|
||
if not sd or not ed:
|
||
continue
|
||
d = (ed - sd).total_seconds() / 60.0 # 分钟
|
||
durations.append(d)
|
||
if d < 10:
|
||
anomalies_short.append((r, d))
|
||
elif d > 20:
|
||
anomalies_long.append((r, d))
|
||
if not durations:
|
||
return dict(count=0, mean=0, median=0, min=0, max=0,
|
||
anomalies_short=[], anomalies_long=[])
|
||
srt = sorted(durations)
|
||
n = len(srt)
|
||
med = srt[n//2] if n%2==1 else (srt[n//2-1]+srt[n//2])/2
|
||
return dict(count=n, mean=round(sum(srt)/n, 1), median=round(med, 1),
|
||
min=round(srt[0], 1), max=round(srt[-1], 1),
|
||
anomalies_short=anomalies_short, anomalies_long=anomalies_long)
|
||
|
||
|
||
def analyze_mid_interaction(comp_records, chapter_map):
|
||
"""分析中互动情况"""
|
||
if not comp_records:
|
||
return dict(total=0, perfect=0, good=0, oops=0, failed=0, pass_cnt=0, none_cnt=0,
|
||
by_chapter={}, trend_chapters=[])
|
||
counts = defaultdict(int)
|
||
ch_data = defaultdict(lambda: {"perfect": 0, "good": 0, "oops": 0, "total": 0})
|
||
chapter_order = []
|
||
seen = set()
|
||
for r in comp_records:
|
||
pr = r.get("play_result", "None")
|
||
counts[pr] = counts.get(pr, 0) + 1
|
||
cid = r.get("chapter_id")
|
||
if cid:
|
||
if cid not in seen:
|
||
seen.add(cid)
|
||
chapter_order.append(cid)
|
||
ch_data[cid]["total"] += 1
|
||
if pr == "Perfect":
|
||
ch_data[cid]["perfect"] += 1
|
||
elif pr == "Good":
|
||
ch_data[cid]["good"] += 1
|
||
elif pr == "Oops":
|
||
ch_data[cid]["oops"] += 1
|
||
total = sum(counts.values())
|
||
# By chapter for trend
|
||
trend = []
|
||
for cid in chapter_order:
|
||
d = ch_data[cid]
|
||
if d["total"] == 0: continue
|
||
ci = chapter_map.get(cid, {})
|
||
trend.append(dict(
|
||
chapter_id=cid, level=ci.get("level",""), unit=ci.get("unit_index",-1),
|
||
lesson=ci.get("lesson_index", 0),
|
||
total=d["total"], perfect=d["perfect"], good=d["good"], oops=d["oops"],
|
||
perfect_pct=round(d["perfect"]/d["total"]*100,1),
|
||
good_pct=round(d["good"]/d["total"]*100,1),
|
||
oops_pct=round(d["oops"]/d["total"]*100,1),
|
||
))
|
||
return dict(total=total, perfect=counts.get("Perfect",0), good=counts.get("Good",0),
|
||
oops=counts.get("Oops",0), failed=counts.get("Failed",0),
|
||
pass_cnt=counts.get("Pass",0), none_cnt=counts.get("None",0),
|
||
trend_chapters=trend)
|
||
|
||
|
||
def analyze_practice(settlements):
|
||
"""分析知识巩固情况"""
|
||
if not settlements:
|
||
return dict(total_lessons=0, completed=0, rate=0,
|
||
avg_score=0, records=[])
|
||
records = []
|
||
completed = 0
|
||
scores = []
|
||
for s in settlements:
|
||
sd = s.get("sd_parsed", {})
|
||
if not sd:
|
||
continue
|
||
prac_count = sd.get("practiceCount", 0) or 0
|
||
prac_score = sd.get("practiceSkillPoint", 0) or 0
|
||
kp_count = sd.get("knowledgePointCount", 0) or 0
|
||
# isPractice 字段不可靠（始终为 false），改用 practiceScore > 0 判断
|
||
has_prac = prac_score > 0
|
||
if has_prac:
|
||
completed += 1
|
||
pct = round(prac_score / 100 * 100, 1) if prac_score > 0 else 0
|
||
scores.append(pct)
|
||
records.append(dict(
|
||
chapter_id=s.get("chapter_id"),
|
||
updated_at=s.get("updated_at"),
|
||
is_practice=has_prac,
|
||
practice_count=prac_count,
|
||
practice_score=prac_score,
|
||
practice_pct=round(prac_score/100*100,1) if prac_score > 0 else 0,
|
||
kp_count=kp_count,
|
||
))
|
||
total = len(records)
|
||
return dict(total_lessons=total, completed=completed,
|
||
completion_rate=round(completed/total*100,1) if total>0 else 0,
|
||
avg_score=round(sum(scores)/len(scores),1) if scores else 0,
|
||
records=records)
|
||
|
||
|
||
def _fmt_duration(mins):
|
||
if mins is None or mins < 0: return "-"
|
||
m = int(mins)
|
||
s = int((mins - m) * 60)
|
||
return f"{m}分{s}秒" if m > 0 else f"{s}秒"
|
||
|
||
|
||
def _td(dt):
|
||
return dt.strftime('%Y-%m-%d %H:%M') if dt else "-"
|
||
|
||
|
||
# ── Markdown ──
|
||
|
||
def format_markdown(role_id, role_info, account_info, course_detail, device_info,
|
||
play_records, dur_analysis, mid_analysis, prac_analysis):
|
||
cm = fetch_chapter_info_map()
|
||
ns = datetime.now().strftime('%Y-%m-%d %H:%M')
|
||
L = []
|
||
|
||
ri = role_info or {}
|
||
ai = account_info or {}
|
||
cd = course_detail or {}
|
||
di = device_info or {}
|
||
|
||
L.append(f"# 📊 上课情况分析报告 — 角色 {role_id}（{ri.get('nickname','')}）\n")
|
||
|
||
# ── 第一步：基础信息 ──
|
||
L.append("## 一、基础信息\n")
|
||
L.append("| 项目 | 详情 |")
|
||
L.append("|------|------|")
|
||
L.append(f"| 角色ID | {role_id} |")
|
||
L.append(f"| 角色姓名 | {ri.get('nickname','-')} |")
|
||
if ri.get('age'): L.append(f"| 年龄 | {ri['age']} 岁 |")
|
||
L.append(f"| 账号ID | {ri.get('account_id','-')} |")
|
||
L.append(f"| 手机号后4位 | {ai.get('phone_tail','-')} |")
|
||
L.append(f"| 账号注册时间 | {ai.get('account_created','-')} |")
|
||
|
||
# 课程购买时间
|
||
act_time = cd.get("active_time")
|
||
exp_time = cd.get("expire_time")
|
||
L.append(f"| 课程激活时间 | {_td(act_time)}" if act_time else "| 课程激活时间 | - |")
|
||
L.append(f"| 课程到期时间 | {_td(exp_time)}" if exp_time else "| 课程到期时间 | - |")
|
||
L.append(f"| 课程级别 | {cd.get('course_level','-')} |")
|
||
L.append(f"| 购买渠道 | {ai.get('download_channel','-')} |")
|
||
L.append(f"| 渠道来源 | {ai.get('key_from','-')} |")
|
||
|
||
# 设备信息
|
||
dev_parts = []
|
||
if di.get("device_name"): dev_parts.append(f"{di['device_name']}")
|
||
if di.get("device_model"): dev_parts.append(f"{di['device_model']}")
|
||
if not dev_parts: dev_parts.append("-")
|
||
L.append(f"| 设备名称 | {' / '.join(dev_parts)} |")
|
||
L.append(f"| 操作系统 | {di.get('os_info','-')[:80]} |")
|
||
L.append(f"| 设备类型 | {di.get('device_type','-')} |")
|
||
L.append(f"| 城市 | {di.get('city','-')} |")
|
||
|
||
# 完课信息
|
||
if play_records:
|
||
first = play_records[0]["updated_at"]
|
||
last = play_records[-1]["updated_at"]
|
||
L.append(f"| 第一次完课 | {_td(first)} |")
|
||
L.append(f"| 最后一次完课 | {_td(last)} |")
|
||
lcid = play_records[-1].get("chapter_id")
|
||
lci = cm.get(lcid, {})
|
||
L.append(f"| 最后一次完课位置 | Level[{lci.get('level','-')}] Unit[{lci.get('unit_index','-')}] Lesson[{lci.get('lesson_index','-')}] |")
|
||
L.append("")
|
||
|
||
L.append(f"**分析时间**: {ns}")
|
||
L.append(f"**完课记录总数**: {len(play_records)} 条")
|
||
L.append(f"**中互动记录总数**: {mid_analysis.get('total',0)} 条")
|
||
L.append(f"**知识巩固记录总数**: {prac_analysis.get('total_lessons',0)} 条\n")
|
||
|
||
# ── 第二步：完课耗时分析 ──
|
||
L.append("---\n## 二、完课耗时分析\n")
|
||
da = dur_analysis
|
||
if da["count"] == 0:
|
||
L.append("> ⚠️ 无有效完课记录。\n")
|
||
else:
|
||
L.append(f"- 有效完课记录: **{da['count']}** 条")
|
||
L.append(f"- 平均完成时间: **{da['mean']}** 分钟")
|
||
L.append(f"- 中位数完成时间: **{da['median']}** 分钟")
|
||
L.append(f"- 最快: **{da['min']}** 分钟 | 最慢: **{da['max']}** 分钟\n")
|
||
|
||
if da["anomalies_short"]:
|
||
L.append(f"### ⚡ 过快完成（<10分钟，共 {len(da['anomalies_short'])} 条）\n")
|
||
L.append("| 日期 | 课程Level/Unit/Lesson | 耗时 |")
|
||
L.append("|------|-----------------------|------|")
|
||
for r, d in da["anomalies_short"][:20]:
|
||
cid = r.get("chapter_id")
|
||
ci = cm.get(cid, {})
|
||
L.append(f"| {_td(r['updated_at'])} | {ci.get('level','')}-U{ci.get('unit_index','')}-L{ci.get('lesson_index','')} | {_fmt_duration(d)} |")
|
||
if len(da["anomalies_short"]) > 20:
|
||
L.append(f"| ... | 等共 {len(da['anomalies_short'])} 条 | |")
|
||
L.append("")
|
||
|
||
if da["anomalies_long"]:
|
||
L.append(f"### 🐢 过慢完成（>20分钟，共 {len(da['anomalies_long'])} 条）\n")
|
||
L.append("| 日期 | 课程Level/Unit/Lesson | 耗时 |")
|
||
L.append("|------|-----------------------|------|")
|
||
for r, d in da["anomalies_long"][:20]:
|
||
cid = r.get("chapter_id")
|
||
ci = cm.get(cid, {})
|
||
L.append(f"| {_td(r['updated_at'])} | {ci.get('level','')}-U{ci.get('unit_index','')}-L{ci.get('lesson_index','')} | {_fmt_duration(d)} |")
|
||
if len(da["anomalies_long"]) > 20:
|
||
L.append(f"| ... | 等共 {len(da['anomalies_long'])} 条 | |")
|
||
L.append("")
|
||
|
||
# 趋势
|
||
if da["count"] >= 5:
|
||
L.append("### 耗时趋势分析")
|
||
# 简单前半/后半对比
|
||
mid = da["count"] // 2
|
||
half1 = sorted([(r["updated_at"] - r["created_at"]).total_seconds()/60
|
||
for r in play_records[:mid] if r.get("created_at") and r.get("updated_at")])
|
||
half2 = sorted([(r["updated_at"] - r["created_at"]).total_seconds()/60
|
||
for r in play_records[mid:] if r.get("created_at") and r.get("updated_at")])
|
||
m1 = round(sum(half1)/len(half1),1) if half1 else 0
|
||
m2 = round(sum(half2)/len(half2),1) if half2 else 0
|
||
if m1 > 0:
|
||
change = round((m2-m1)/m1*100,1)
|
||
trend_desc = "缩短" if change < -5 else ("延长" if change > 5 else "基本持平")
|
||
L.append(f"- 前半段平均耗时: **{m1}** 分钟 → 后半段: **{m2}** 分钟（{trend_desc} {abs(change)}%）")
|
||
L.append("")
|
||
|
||
# ── 第三步：中互动分析 ──
|
||
L.append("---\n## 三、中互动正确率分析\n")
|
||
ma = mid_analysis
|
||
if ma.get("total", 0) == 0:
|
||
L.append("> ⚠️ 无中互动记录。\n")
|
||
else:
|
||
t = ma["total"]
|
||
perfect_pct = round(ma["perfect"]/t*100,1)
|
||
good_pct = round(ma["good"]/t*100,1)
|
||
oops_pct = round(ma["oops"]/t*100,1)
|
||
failed_pct = round(ma.get("failed",0)/t*100,1)
|
||
pass_pct = round(ma.get("pass_cnt",0)/t*100,1)
|
||
|
||
L.append(f"| 结果 | 数量 | 占比 |")
|
||
L.append(f"|------|------|------|")
|
||
L.append(f"| Perfect ✨ | {ma['perfect']} | {perfect_pct}% |")
|
||
L.append(f"| Good 👍 | {ma['good']} | {good_pct}% |")
|
||
L.append(f"| Oops 😅 | {ma['oops']} | {oops_pct}% |")
|
||
if ma.get("pass_cnt", 0) > 0:
|
||
L.append(f"| Pass | {ma['pass_cnt']} | {pass_pct}% |")
|
||
if ma.get("failed", 0) > 0:
|
||
L.append(f"| Failed | {ma['failed']} | {failed_pct}% |")
|
||
L.append("")
|
||
|
||
# 优良率
|
||
excellent = perfect_pct + good_pct
|
||
L.append(f"**优良率（Perfect+Good）**: {excellent}%\n")
|
||
|
||
# 趋势
|
||
if ma.get("trend_chapters"):
|
||
mtr = ma["trend_chapters"]
|
||
# Split in half
|
||
mh = len(mtr)//2
|
||
h1_p = [c["perfect_pct"] for c in mtr[:mh]]
|
||
h2_p = [c["perfect_pct"] for c in mtr[mh:]]
|
||
h1_avg = round(sum(h1_p)/len(h1_p),1) if h1_p else 0
|
||
h2_avg = round(sum(h2_p)/len(h2_p),1) if h2_p else 0
|
||
if h1_avg > 0:
|
||
trend = "上升" if h2_avg > h1_avg + 5 else ("下降" if h2_avg < h1_avg - 5 else "持平")
|
||
L.append(f"**趋势**: 前半段平均 Perfect 率 {h1_avg}% → 后半段 {h2_avg}%（{trend}）")
|
||
L.append("")
|
||
|
||
# ── 第四步：巩固题分析 ──
|
||
L.append("---\n## 四、知识巩固分析\n")
|
||
pa = prac_analysis
|
||
if pa.get("total_lessons", 0) == 0:
|
||
L.append("> ⚠️ 无知识巩固记录。\n")
|
||
else:
|
||
L.append(f"- 总课程数: **{pa['total_lessons']}** 节")
|
||
L.append(f"- 完成巩固练习: **{pa['completed']}** 节（完成率 **{pa['completion_rate']}%**）")
|
||
L.append(f"- 巩固题平均得分: **{pa['avg_score']}**%（满分100）\n")
|
||
|
||
# 未完成的课程
|
||
skipped = [r for r in pa.get("records", []) if not r["is_practice"]]
|
||
if skipped:
|
||
L.append(f"### 未完成巩固练习的课程（共 {len(skipped)} 节）\n")
|
||
L.append("| 课程 | 日期 |")
|
||
L.append("|------|------|")
|
||
for r in skipped[:15]:
|
||
cid = r["chapter_id"]
|
||
ci = cm.get(cid, {})
|
||
L.append(f"| {ci.get('level','')}-U{ci.get('unit_index','')}-L{ci.get('lesson_index','')} | {_td(r['updated_at'])} |")
|
||
if len(skipped) > 15:
|
||
L.append(f"| ... | 等共 {len(skipped)} 节 |")
|
||
L.append("")
|
||
|
||
# 得分分布
|
||
scores = [r["practice_score"] for r in pa.get("records", []) if r["is_practice"]]
|
||
if scores:
|
||
high = sum(1 for s in scores if s >= 80)
|
||
mid_s = sum(1 for s in scores if 50 <= s < 80)
|
||
low = sum(1 for s in scores if s < 50)
|
||
L.append(f"### 巩固题得分分布")
|
||
L.append(f"- 高分（≥80分）: **{high}** 节（{round(high/len(scores)*100,1)}%）")
|
||
L.append(f"- 中等（50-79分）: **{mid_s}** 节（{round(mid_s/len(scores)*100,1)}%）")
|
||
L.append(f"- 低分（<50分）: **{low}** 节（{round(low/len(scores)*100,1)}%）")
|
||
L.append("")
|
||
|
||
L.append("---")
|
||
L.append("> 💡 是否需要将以上详细分析报告生成一份 HTML 文件？回复「是」或「需要」即可。")
|
||
return "\n".join(L)
|
||
|
||
|
||
# ── HTML ──
|
||
|
||
def format_html(role_id, role_info, account_info, course_detail, device_info,
|
||
play_records, dur_analysis, mid_analysis, prac_analysis):
|
||
cm = fetch_chapter_info_map()
|
||
ns = datetime.now().strftime('%Y-%m-%d %H:%M')
|
||
|
||
ri = role_info or {}
|
||
ai = account_info or {}
|
||
cd = course_detail or {}
|
||
di = device_info or {}
|
||
|
||
# Part 1: Basic info
|
||
act_time = cd.get("active_time")
|
||
exp_time = cd.get("expire_time")
|
||
dev_parts = []
|
||
if di.get("device_name"): dev_parts.append(di['device_name'])
|
||
if di.get("device_model"): dev_parts.append(di['device_model'])
|
||
if not dev_parts: dev_parts.append("-")
|
||
|
||
first_dt = last_dt = last_lvl = "-"
|
||
if play_records:
|
||
first_dt = _td(play_records[0]["updated_at"])
|
||
last_dt = _td(play_records[-1]["updated_at"])
|
||
lcid = play_records[-1].get("chapter_id")
|
||
lci = cm.get(lcid, {})
|
||
last_lvl = f"Level[{lci.get('level','-')}] Unit[{lci.get('unit_index','-')}] Lesson[{lci.get('lesson_index','-')}]"
|
||
|
||
# Part 2: Detail table
|
||
rows_html = ""
|
||
prac_dict = {r["chapter_id"]: r for r in prac_analysis.get("records", [])}
|
||
ma_trend = {t["chapter_id"]: t for t in mid_analysis.get("trend_chapters", [])}
|
||
for i, pr in enumerate(play_records, 1):
|
||
cid = pr.get("chapter_id")
|
||
ci = cm.get(cid, {})
|
||
sd = pr.get("created_at")
|
||
ed = pr["updated_at"]
|
||
dur = (ed - sd).total_seconds() / 60 if sd and ed else None
|
||
|
||
# Mid-interaction info
|
||
mt = ma_trend.get(cid, {})
|
||
mid_str = f"Perfect[{mt.get('perfect',0)},{mt.get('perfect_pct',0)}%]-Good[{mt.get('good',0)},{mt.get('good_pct',0)}%]-Oops[{mt.get('oops',0)},{mt.get('oops_pct',0)}%]" if mt else "-"
|
||
|
||
# Practice info
|
||
pd = prac_dict.get(cid)
|
||
prac_done = "是" if pd and pd.get("is_practice") else "否"
|
||
prac_cnt = pd.get("practice_count", "-") if pd else "-"
|
||
prac_pct = f"{pd.get('practice_pct','-')}%" if pd and pd.get("is_practice") else "-"
|
||
|
||
rows_html += f"""<tr>
|
||
<td>{ci.get('level','')}-U{ci.get('unit_index','')}-L{ci.get('lesson_index','')}</td>
|
||
<td>{_td(sd)}</td><td>{_td(ed)}</td><td>{_fmt_duration(dur)}</td>
|
||
<td>{mid_str}</td><td>{prac_done}</td><td>{prac_cnt}</td><td>{prac_pct}</td></tr>"""
|
||
|
||
# Part 3: Duration analysis
|
||
da = dur_analysis
|
||
dur_summary = "无有效完课记录。"
|
||
if da["count"] > 0:
|
||
trend_text = ""
|
||
if da["count"] >= 5:
|
||
mid = da["count"] // 2
|
||
half1 = sorted([(r["updated_at"] - r["created_at"]).total_seconds()/60
|
||
for r in play_records[:mid] if r.get("created_at") and r.get("updated_at")])
|
||
half2 = sorted([(r["updated_at"] - r["created_at"]).total_seconds()/60
|
||
for r in play_records[mid:] if r.get("created_at") and r.get("updated_at")])
|
||
m1 = round(sum(half1)/len(half1),1) if half1 else 0
|
||
m2 = round(sum(half2)/len(half2),1) if half2 else 0
|
||
if m1 > 0:
|
||
change = round((m2-m1)/m1*100,1)
|
||
trend = "缩短" if change < -5 else ("延长" if change > 5 else "持平")
|
||
trend_text = f"趋势：前半段平均 {m1} 分钟 → 后半段 {m2} 分钟（{trend} {abs(change)}%）。"
|
||
dur_summary = (f"共 {da['count']} 条有效记录，平均 {da['mean']} 分钟，"
|
||
f"中位数 {da['median']} 分钟。最快 {da['min']} 分钟，最慢 {da['max']} 分钟。"
|
||
f"异常：过快 {len(da['anomalies_short'])} 条（<10分钟），"
|
||
f"过慢 {len(da['anomalies_long'])} 条（>20分钟）。{trend_text}")
|
||
|
||
# Part 4: Mid-interaction analysis
|
||
ma = mid_analysis
|
||
mid_text = "无中互动记录。"
|
||
if ma.get("total", 0) > 0:
|
||
t = ma["total"]
|
||
p_pct = round(ma["perfect"]/t*100,1)
|
||
g_pct = round(ma["good"]/t*100,1)
|
||
o_pct = round(ma["oops"]/t*100,1)
|
||
exc = p_pct + g_pct
|
||
trend_text = ""
|
||
if ma.get("trend_chapters"):
|
||
mtr = ma["trend_chapters"]
|
||
mh = len(mtr)//2
|
||
h1_p = [c["perfect_pct"] for c in mtr[:mh]] if mtr[:mh] else [0]
|
||
h2_p = [c["perfect_pct"] for c in mtr[mh:]] if mtr[mh:] else [0]
|
||
h1_avg = round(sum(h1_p)/len(h1_p),1)
|
||
h2_avg = round(sum(h2_p)/len(h2_p),1)
|
||
if h1_avg > 0:
|
||
trend = "上升" if h2_avg > h1_avg + 5 else ("下降" if h2_avg < h1_avg - 5 else "持平")
|
||
trend_text = f"趋势：Perfect 率 {h1_avg}% → {h2_avg}%（{trend}）。"
|
||
mid_text = (f"共 {t} 条记录。Perfect {p_pct}%，Good {g_pct}%，Oops {o_pct}%，"
|
||
f"优良率 {exc}%。{trend_text}")
|
||
|
||
# Part 5: Practice analysis
|
||
pa = prac_analysis
|
||
prac_text = "无知识巩固记录。"
|
||
if pa.get("total_lessons", 0) > 0:
|
||
scores = [r["practice_score"] for r in pa.get("records", []) if r["is_practice"]]
|
||
score_dist = ""
|
||
if scores:
|
||
high = sum(1 for s in scores if s >= 80)
|
||
mid_s = sum(1 for s in scores if 50 <= s < 80)
|
||
low = sum(1 for s in scores if s < 50)
|
||
score_dist = f"得分分布：高分（≥80）{high}节 / 中等 {mid_s}节 / 低分（<50）{low}节。"
|
||
prac_text = (f"共 {pa['total_lessons']} 节课，完成巩固练习 {pa['completed']} 节"
|
||
f"（完成率 {pa['completion_rate']}%），平均得分 {pa['avg_score']}%。{score_dist}")
|
||
|
||
# Summary
|
||
summary_items = []
|
||
if da["count"] > 0:
|
||
if da["mean"] < 15:
|
||
summary_items.append("⚡ 完课速度较快，平均不足15分钟，建议关注是否有跳课现象")
|
||
elif da["mean"] > 20:
|
||
summary_items.append("🐢 完课耗时较长，建议关注学习效率")
|
||
summary_items.append(f"📊 完课耗时中位数 {da['median']} 分钟")
|
||
if ma.get("total", 0) > 0:
|
||
exc = round(ma["perfect"]/ma["total"]*100 + ma["good"]/ma["total"]*100,1)
|
||
summary_items.append(f"🎯 中互动优良率 {exc}%")
|
||
if pa.get("total_lessons", 0) > 0:
|
||
summary_items.append(f"📝 知识巩固完成率 {pa['completion_rate']}%，平均得分 {pa['avg_score']}%")
|
||
si_html = "".join(f"<li>{it}</li>" for it in summary_items) if summary_items else "<li>暂无分析数据</li>"
|
||
|
||
# Precompute values for HTML template (avoid f-string concatenation)
|
||
age_str = f"{ri['age']} 岁" if ri.get('age') else '-'
|
||
dev_str = '/'.join(dev_parts)
|
||
os_str = (di.get('os_info','-') or '-')[:80]
|
||
|
||
return f"""<!DOCTYPE html>
|
||
<html lang="zh-CN">
|
||
<head>
|
||
<meta charset="UTF-8">
|
||
<meta name="viewport" content="width=device-width,initial-scale=1.0">
|
||
<title>上课情况分析 — 角色 {role_id}</title>
|
||
<style>
|
||
*{{margin:0;padding:0;box-sizing:border-box}}
|
||
body{{font-family:-apple-system,BlinkMacSystemFont,"Segoe UI","PingFang SC","Microsoft YaHei",sans-serif;background:#f5f7fa;color:#333;line-height:1.6;padding:20px}}
|
||
.container{{max-width:1200px;margin:0 auto}}
|
||
h1{{text-align:center;color:#1a1a2e;margin:20px 0 30px;font-size:24px}}
|
||
h2{{color:#2d3436;font-size:20px;margin:30px 0 15px;padding-bottom:8px;border-bottom:2px solid #0984e3}}
|
||
.card{{background:#fff;border-radius:10px;box-shadow:0 2px 8px rgba(0,0,0,.08);padding:24px;margin-bottom:20px}}
|
||
.info-grid{{display:grid;grid-template-columns:repeat(auto-fill,minmax(280px,1fr));gap:12px}}
|
||
.info-item{{display:flex;padding:8px 0;border-bottom:1px solid #f0f0f0}}
|
||
.info-label{{color:#636e72;min-width:140px;font-weight:500}}
|
||
.info-value{{color:#2d3436}}
|
||
table{{width:100%;border-collapse:collapse;font-size:12px}}
|
||
thead{{background:#0984e3;color:#fff}}
|
||
th{{padding:10px 6px;text-align:center;font-weight:600;white-space:nowrap}}
|
||
td{{padding:8px 6px;text-align:center;border-bottom:1px solid #eee;font-size:11px}}
|
||
tr:hover td{{background:#f0f7ff}}
|
||
.summary-list{{padding-left:20px}}
|
||
.summary-list li{{margin-bottom:8px;line-height:1.8}}
|
||
.meta{{text-align:center;color:#999;font-size:12px;margin-top:30px}}
|
||
.text-block{{background:#f8f9fd;border-left:4px solid #0984e3;padding:12px 16px;border-radius:0 8px 8px 0;margin:10px 0}}
|
||
@media print{{body{{background:#fff}}.card{{box-shadow:none;border:1px solid #ddd}}table{{font-size:10px}}td,th{{padding:6px 4px}}}}
|
||
</style></head>
|
||
<body><div class="container">
|
||
|
||
<h1>📊 上课情况分析报告 — 角色 {role_id}（{ri.get('nickname','')}）</h1>
|
||
|
||
<h2>一、基础信息</h2>
|
||
<div class="card"><div class="info-grid">
|
||
<div class="info-item"><span class="info-label">角色ID</span><span class="info-value">{role_id}</span></div>
|
||
<div class="info-item"><span class="info-label">角色姓名</span><span class="info-value">{ri.get('nickname','-')}</span></div>
|
||
<div class="info-item"><span class="info-label">年龄</span><span class="info-value">{age_str}</span></div>
|
||
<div class="info-item"><span class="info-label">账号ID</span><span class="info-value">{ri.get('account_id','-')}</span></div>
|
||
<div class="info-item"><span class="info-label">手机号后4位</span><span class="info-value">{ai.get('phone_tail','-')}</span></div>
|
||
<div class="info-item"><span class="info-label">账号注册时间</span><span class="info-value">{ai.get('account_created','-')}</span></div>
|
||
<div class="info-item"><span class="info-label">课程激活时间</span><span class="info-value">{_td(act_time)}</span></div>
|
||
<div class="info-item"><span class="info-label">课程到期时间</span><span class="info-value">{_td(exp_time)}</span></div>
|
||
<div class="info-item"><span class="info-label">课程级别</span><span class="info-value">{cd.get('course_level','-')}</span></div>
|
||
<div class="info-item"><span class="info-label">购买渠道</span><span class="info-value">{ai.get('download_channel','-')}</span></div>
|
||
<div class="info-item"><span class="info-label">渠道来源</span><span class="info-value">{ai.get('key_from','-')}</span></div>
|
||
<div class="info-item"><span class="info-label">设备名称</span><span class="info-value">{dev_str}</span></div>
|
||
<div class="info-item"><span class="info-label">操作系统</span><span class="info-value">{os_str}</span></div>
|
||
<div class="info-item"><span class="info-label">设备类型</span><span class="info-value">{di.get('device_type','-')}</span></div>
|
||
<div class="info-item"><span class="info-label">城市</span><span class="info-value">{di.get('city','-')}</span></div>
|
||
<div class="info-item"><span class="info-label">第一次完课</span><span class="info-value">{first_dt}</span></div>
|
||
<div class="info-item"><span class="info-label">最后一次完课</span><span class="info-value">{last_dt}</span></div>
|
||
<div class="info-item"><span class="info-label">最后完课位置</span><span class="info-value">{last_lvl}</span></div>
|
||
</div></div>
|
||
|
||
<h2>二、完课记录明细</h2>
|
||
<div class="card" style="overflow-x:auto">
|
||
<table><thead><tr>
|
||
<th>课程</th><th>开始时间</th><th>完课时间</th><th>耗时</th><th>中互动</th><th>巩固完成</th><th>巩固题数</th><th>巩固正确率</th>
|
||
</tr></thead><tbody>
|
||
{rows_html}
|
||
</tbody></table></div>
|
||
|
||
<h2>三、上课时间分析</h2>
|
||
<div class="card"><div class="text-block">{dur_summary}</div></div>
|
||
|
||
<h2>四、中互动正确率分析</h2>
|
||
<div class="card"><div class="text-block">{mid_text}</div></div>
|
||
|
||
<h2>五、知识巩固正确率分析</h2>
|
||
<div class="card"><div class="text-block">{prac_text}</div></div>
|
||
|
||
<h2>总结</h2>
|
||
<div class="card"><ul class="summary-list">{si_html}</ul></div>
|
||
|
||
<div class="meta">分析时间：{ns} | 完课：{len(play_records)} 条 | 中互动：{mid_analysis.get('total',0)} 条 | 巩固：{prac_analysis.get('total_lessons',0)} 条</div>
|
||
|
||
</div></body></html>"""
|
||
|
||
|
||
# ── main ──
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="角色上课情况分析工具")
|
||
ap.add_argument("role_id", type=int)
|
||
ap.add_argument("--format", choices=["md","html"], default="md")
|
||
ap.add_argument("--output","-o", default=None)
|
||
args = ap.parse_args()
|
||
|
||
role_id = args.role_id
|
||
|
||
ri = fetch_role_info(role_id)
|
||
if not ri:
|
||
print(f"角色 {role_id} 不存在")
|
||
sys.exit(1)
|
||
|
||
ai = fetch_account_info(ri["account_id"])
|
||
cd = fetch_course_detail(role_id)
|
||
di = fetch_device_info(ri["account_id"])
|
||
pr = fetch_play_records(role_id)
|
||
cr = fetch_component_records(role_id)
|
||
sd = fetch_settlement_data(role_id)
|
||
|
||
da = analyze_duration(pr)
|
||
ma = analyze_mid_interaction(cr, fetch_chapter_info_map())
|
||
pa = analyze_practice(sd)
|
||
|
||
if args.format == "html":
|
||
out = format_html(role_id, ri, ai, cd, di, pr, da, ma, pa)
|
||
if args.output:
|
||
os.makedirs(os.path.dirname(args.output) or ".", exist_ok=True)
|
||
with open(args.output, "w", encoding="utf-8") as f:
|
||
f.write(out)
|
||
print(f"HTML 报告已保存到: {args.output}")
|
||
else:
|
||
print(out)
|
||
else:
|
||
print(format_markdown(role_id, ri, ai, cd, di, pr, da, ma, pa))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|