# -*- coding: utf-8 -*- """ 家庭练习 · 后端自动回传服务（纯标准库，零依赖） - 静态托管各练习页 / DOCX - POST /api/submit 单题结果上报（page 隔离；对/错） - POST /api/submit action=report 题目报错（答案/题面问题） - GET /api/results 全量结果（按卷分组） - GET /api/reports 报错清单 - GET /api/history 按日汇总做题记录（{date:{page:{count,correct,items}}}） 存储：results.json = {page: {no: rec}}；reports.json = [ {...} ] """ import json, os, shutil, threading from datetime import datetime # 公网鉴权 token：环境变量优先，其次同目录 token.txt；两者都没有 = 不校验（本地开发态） AUTH_TOKEN = os.environ.get('AUTH_TOKEN', '') if not AUTH_TOKEN and os.path.exists(os.path.join(os.path.dirname(os.path.abspath(__file__)), 'token.txt')): AUTH_TOKEN = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), 'token.txt'), encoding='utf-8').read().strip() from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler from datetime import datetime os.chdir(os.path.dirname(os.path.abspath(__file__))) # 静态根 = 本目录 PORT = int(os.environ.get('PORT', 3000)) DATA = os.path.join(os.getcwd(), 'results.json') RPT = os.path.join(os.getcwd(), 'reports.json') LOCK = threading.Lock() def atomic_write(path, obj): """先写临时文件再原子替换，写前保留上一版 .bak。 避免『写到一半进程被杀 → 文件损坏 → load 静默返回空 → 整卷归零』。""" tmp = path + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(obj, f, ensure_ascii=False, indent=1) f.flush() os.fsync(f.fileno()) # 两级轮换备份：当前版→.bak，原 .bak→.bak2。 # 单级备份扛不住「清空等破坏性操作 + 之后写入损坏」叠加，会把备份也带空。 try: if os.path.exists(path + '.bak'): shutil.copyfile(path + '.bak', path + '.bak2') if os.path.exists(path): shutil.copyfile(path, path + '.bak') except Exception: pass os.replace(tmp, path) def _parse(d): if not isinstance(d, dict): return None # 兼容旧扁平结构 {no: rec} → 迁到 'legacy' if d and all(not isinstance(v, dict) or ('state' in v) for v in d.values()): return {'legacy': d} return d def load_results(): """主文件损坏时回退 .bak，而不是静默返回空导致全卷丢失。""" for p in (DATA, DATA + '.bak', DATA + '.bak2'): try: d = _parse(json.load(open(p, encoding='utf-8'))) if d is not None: return d except Exception: continue return {} def load_reports(): try: return json.load(open(RPT, encoding='utf-8')) except Exception: return [] def build_history(res): """按日期汇总做题记录：{date: {page: {count, correct, items:[{no,subject,kp,pick,state,hhmm}]}}}""" hist = {} for page, qs in res.items(): for no, r in qs.items(): ts = r.get('ts') try: day = datetime.fromtimestamp(int(ts) / 1000).strftime('%Y-%m-%d') hhmm = datetime.fromtimestamp(int(ts) / 1000).strftime('%H:%M') except Exception: day = (r.get('recv_at') or '?')[:10] hhmm = (r.get('recv_at') or '?')[11:16] d = hist.setdefault(day, {}) p = d.setdefault(page, {'count': 0, 'correct': 0, 'items': []}) p['count'] += 1 p['correct'] += 1 if r.get('state') == '对' else 0 p['items'].append({'no': no, 'subject': r.get('subject'), 'kp': r.get('kp'), 'pick': r.get('pick'), 'state': r.get('state'), 'time': hhmm}) # 每天内按页、题号排序 for d in hist.values(): for p in d.values(): p['items'].sort(key=lambda x: (str(x.get('subject') or ''), int(x['no']) if str(x['no']).isdigit() else 0)) return dict(sorted(hist.items(), reverse=True)) class Handler(SimpleHTTPRequestHandler): protocol_version = 'HTTP/1.1' # keep-alive：平台反向代理的预期协议，避免连接池打到半死连接 def end_headers(self): # HTML 不缓存：防止浏览器拿旧版主页/练习页（2026-09-06 学情页"返回旧主页"事故根因） p = self.path.split('?')[0].lower() if p == '/' or p.endswith('.html'): self.send_header('Cache-Control', 'no-cache, no-store, must-revalidate') super().end_headers() def _auth_ok(self): # 公网鉴权：token 来自环境变量 AUTH_TOKEN 或同目录 token.txt（都没有=不校验，本地开发用） global AUTH_TOKEN if not AUTH_TOKEN: return True return self.headers.get('X-Auth-Token', '') == AUTH_TOKEN def _json(self, obj, code=200): b = json.dumps(obj, ensure_ascii=False).encode('utf-8') self.send_response(code) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(b))) self.send_header('Cache-Control', 'no-store') # 允许静态页与后端分离部署（如页面留在原平台、后端迁腾讯云） # X-Auth-Token 必须列入 Allow-Headers，否则浏览器 CORS 预检直接拦掉所有带 token 的请求 self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Headers', 'Content-Type, X-Auth-Token') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Max-Age', '86400') self.end_headers() self.wfile.write(b) def do_OPTIONS(self): self._json({'ok': True}) def do_GET(self): if self.path.startswith('/api/'): if not self._auth_ok(): return self._json({'ok': False, 'err': 'unauthorized'}, 401) if self.path == '/api/results': with LOCK: self._json(load_results()) elif self.path == '/api/reports': with LOCK: self._json(load_reports()) elif self.path == '/api/history': with LOCK: self._json(build_history(load_results())) else: self._json({'ok': False, 'err': 'not found'}, 404) else: super().do_GET() def do_POST(self): if self.path != '/api/submit': return self._json({'ok': False, 'err': 'not found'}, 404) if not self._auth_ok(): return self._json({'ok': False, 'err': 'unauthorized'}, 401) try: n = int(self.headers.get('Content-Length', 0)) payload = json.loads(self.rfile.read(n)) except Exception: return self._json({'ok': False, 'err': 'bad body'}, 400) page = str(payload.get('page') or 'legacy') with LOCK: act = payload.get('action') if act == 'report': rpt = load_reports() rpt.append({ 'page': page, 'no': payload.get('no'), 'note': str(payload.get('note') or '')[:200], 'ts': payload.get('ts'), 'recv_at': datetime.now().isoformat(timespec='seconds'), }) atomic_write(RPT, rpt) elif act == 'sync': # 前端自动恢复：把本地有、服务端缺的记录一次性批量补传 items = payload.get('items') or [] d = load_results() pg = d.setdefault(page, {}) n = 0 for it in items: if not isinstance(it, dict): continue if it.get('no') is None or it.get('state') not in ('对', '错'): continue rec = {k: it[k] for k in ('subject', 'kp', 'diff', 'pick', 'key', 'state', 'ts') if k in it} rec['recv_at'] = datetime.now().isoformat(timespec='seconds') pg[str(it['no'])] = rec n += 1 atomic_write(DATA, d) return self._json({'ok': True, 'synced': n}) elif act == 'clear': # 清空本卷：本机与服务端一并清，避免两边对不上 d = load_results() d[page] = {} atomic_write(DATA, d) return self._json({'ok': True, 'cleared': page}) else: d = load_results() pg = d.setdefault(page, {}) if payload.get('no') and payload.get('state') in ('对', '错'): rec = {k: payload[k] for k in ('subject', 'kp', 'diff', 'pick', 'key', 'state', 'ts') if k in payload} rec['recv_at'] = datetime.now().isoformat(timespec='seconds') pg[str(payload['no'])] = rec else: return self._json({'ok': False, 'err': 'invalid'}, 400) atomic_write(DATA, d) self._json({'ok': True}) def log_message(self, *a): # 静音访问日志 pass if __name__ == '__main__': print(f'serving on 0.0.0.0:{PORT}') ThreadingHTTPServer.request_queue_size = 128 # 默认5，代理并发下会拒连 ThreadingHTTPServer(('0.0.0.0', PORT), Handler).serve_forever()