jack/digital-psychology
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
design-vision-v1
digital-psychology/server.py
T

617 lines
20 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""Digital Psychology server — static files + mindmap + scale library API."""
import json, os, sys, sqlite3, time, uuid
from http.server import HTTPServer, SimpleHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
ROOT = os.path.dirname(os.path.abspath(__file__))
MINDMAP_FILE = os.path.join(ROOT, "mindmap_data.json")
SCALES_FILE = os.path.join(ROOT, "scales_data.json")
RESULTS_DB = os.path.join(ROOT, "scale_results.db")
# ---- SQLite setup ----
def get_db():
db = sqlite3.connect(RESULTS_DB)
db.row_factory = sqlite3.Row
return db
def init_db():
db = get_db()
db.execute("""
CREATE TABLE IF NOT EXISTS results (
id TEXT PRIMARY KEY,
scale_slug TEXT NOT NULL,
scale_name TEXT NOT NULL,
answers TEXT NOT NULL,
scores TEXT,
created_at TEXT NOT NULL,
user_agent TEXT,
ip TEXT
)
""")
db.commit()
db.close()
init_db()
# ============================================================
# Scoring engine — known rules for major scales
# ============================================================
SCORE_SCALE = {
# (option_start, reverse_items, transform, rating_thresholds, factor_map)
# option_start: 0=0-indexed (PHQ/GAD), 1=1-indexed (SDS/SAS/SCL-90)
# reverse_items: 1-indexed question numbers to reverse-score
# transform: (multiplier, round) applied to raw score, or None
# rating_thresholds: [(max_score, label), ...]
# factor_map: {factor_name: [1-indexed question numbers]}
}
SCORING = {
# === 抑郁症测试 ===
"yyzcs": { # SDS 抑郁自评量表
"name": "抑郁自评量表(SDS)",
"option_base": 1, # 1-4
"reverse": [2, 5, 6, 11, 12, 14, 16, 17, 18, 20],
"transform": (1.25, 0), # std = raw * 1.25, rounded
"ratings": [
(52, "可能没有抑郁"),
(62, "可能轻度抑郁"),
(72, "可能中度抑郁"),
(100, "可能重度抑郁"),
],
"factors": {
"精神性情感症状": [1, 3],
"躯体性障碍": [2, 4, 5, 6, 7, 8, 9, 10],
"精神运动性障碍": [12, 13],
"抑郁的心理障碍": [11, 14, 15, 16, 17, 18, 19, 20],
},
"use_std": True,
},
"yyz": { # BDI-II 贝克抑郁量表
"name": "贝克抑郁量表(BDI-II)",
"option_base": 0, # 0-3
"reverse": [],
"transform": None,
"ratings": [
(13, "无抑郁或极轻微"),
(19, "轻度抑郁"),
(28, "中度抑郁"),
(63, "重度抑郁"),
],
"factors": {},
"use_std": False,
},
"yy9": { # 中学生抑郁自评量表
"name": "中学生抑郁自评量表",
"option_base": 1,
"reverse": [],
"transform": None,
"ratings": [
(39, "可能没有抑郁"),
(47, "可能轻度抑郁"),
(55, "可能中度抑郁"),
(80, "可能重度抑郁"),
],
"factors": {},
"use_std": False,
},
"yy4": { # CES-D 流调用抑郁量表
"name": "流调用抑郁量表(CES-D)",
"option_base": 0,
"reverse": [4, 8, 12, 16],
"transform": None,
"ratings": [
(15, "无抑郁症状"),
(19, "可能有抑郁倾向"),
(24, "可能有抑郁症状"),
(60, "可能有严重抑郁症状"),
],
"factors": {},
"use_std": False,
},
"yy6": { # PHQ-9
"name": "抑郁症筛查量表(PHQ-9)",
"option_base": 0,
"reverse": [],
"transform": None,
"ratings": [
(4, "无抑郁"),
(9, "轻度抑郁"),
(14, "中度抑郁"),
(19, "中重度抑郁"),
(27, "重度抑郁"),
],
"factors": {},
"use_std": False,
},
"yy7": { # CDI 儿童青少年抑郁量表
"name": "儿童青少年抑郁量表(CDI)",
"option_base": 0,
"reverse": [],
"transform": None,
"ratings": [
(19, "无抑郁"),
(29, "轻度抑郁"),
(39, "中度抑郁"),
(54, "重度抑郁"),
],
"factors": {},
"use_std": False,
},
"yy8": { # DSRSC 儿童抑郁障碍自评
"name": "儿童抑郁障碍自评量表(DSRSC)",
"option_base": 0,
"reverse": [],
"transform": None,
"ratings": [
(13, "正常"),
(17, "可能有抑郁"),
(36, "很可能有抑郁"),
],
"factors": {},
"use_std": False,
},
"yy10": { # GDS 老年抑郁量表
"name": "老年抑郁量表(GDS)",
"option_base": 0,
"reverse": [1, 5, 7, 9, 15, 19, 21, 27, 29, 30],
"transform": None,
"ratings": [
(10, "正常"),
(19, "轻度抑郁"),
(30, "重度抑郁"),
],
"factors": {},
"use_std": False,
},
"yy11": { # EPDS 爱丁堡产后抑郁
"name": "爱丁堡产后抑郁量表(EPDS)",
"option_base": 0,
"reverse": [],
"transform": None,
"ratings": [
(9, "正常"),
(12, "可能存在抑郁"),
(30, "很可能存在抑郁"),
],
"factors": {},
"use_std": False,
},
"yy12": { # HAD 焦虑抑郁量表
"name": "焦虑抑郁量表(HAD)",
"option_base": 0,
"reverse": [],
"transform": None,
"ratings": [
(7, "正常"),
(10, "临界"),
(21, "异常"),
],
"factors": {},
"use_std": False,
},
"yy13": { # HAMD 汉密尔顿抑郁
"name": "汉密尔顿抑郁量表(HAMD)",
"option_base": 0,
"reverse": [],
"transform": None,
"ratings": [
(7, "正常"),
(17, "可能有轻中度抑郁"),
(24, "可能有重度抑郁"),
(72, "可能有极重度抑郁"),
],
"factors": {},
"use_std": False,
},
# === 焦虑症测试 ===
"jl1": { # SAS 焦虑自评量表
"name": "焦虑自评量表(SAS)",
"option_base": 1,
"reverse": [5, 9, 13, 17, 19],
"transform": (1.25, 0),
"ratings": [
(49, "可能没有焦虑"),
(59, "可能轻度焦虑"),
(69, "可能中度焦虑"),
(100, "可能重度焦虑"),
],
"factors": {},
"use_std": True,
},
"jlz": { # BAI 贝克焦虑量表
"name": "贝克焦虑测试量表(BAI)",
"option_base": 0,
"reverse": [],
"transform": None,
"ratings": [
(7, "无焦虑"),
(15, "轻度焦虑"),
(25, "中度焦虑"),
(63, "重度焦虑"),
],
"factors": {},
"use_std": False,
},
"jl5": { # GAD-7
"name": "焦虑症筛查量表(GAD-7)",
"option_base": 0,
"reverse": [],
"transform": None,
"ratings": [
(4, "无焦虑"),
(9, "轻度焦虑"),
(14, "中度焦虑"),
(21, "重度焦虑"),
],
"factors": {},
"use_std": False,
},
"jl6": { # TAS 考试焦虑
"name": "考试焦虑量表(TAS)",
"option_base": 0,
"reverse": [],
"transform": None,
"ratings": [
(12, "较低考试焦虑"),
(20, "中等考试焦虑"),
(37, "较高考试焦虑"),
],
"factors": {},
"use_std": False,
},
"jl8": { # 中学生焦虑自评
"name": "中学生焦虑自评量表",
"option_base": 1,
"reverse": [],
"transform": None,
"ratings": [
(39, "可能没有焦虑"),
(48, "可能轻度焦虑"),
(56, "可能中度焦虑"),
(80, "可能重度焦虑"),
],
"factors": {},
"use_std": False,
},
# === 强迫症测试 ===
"qp1": { # YBOCS
"name": "耶鲁布朗强迫标准量表(YBOCS)",
"option_base": 0,
"reverse": [],
"transform": None,
"ratings": [
(7, "亚临床"),
(15, "轻度"),
(23, "中度"),
(31, "重度"),
(40, "极重度"),
],
"factors": {},
"use_std": False,
},
# === 人格测试 ===
"xl16": { # NEO-FFI 大五人格
"name": "大五人格测试(NEO-FFI)",
"option_base": 1,
"reverse": [], # varies by factor
"transform": None,
"ratings": [], # factor-based
"factors": {
"神经质": [1, 6, 11, 16, 21, 26, 31, 36, 41, 46, 51, 56], # approximate
"外向性": [2, 7, 12, 17, 22, 27, 32, 37, 42, 47, 52, 57],
"开放性": [3, 8, 13, 18, 23, 28, 33, 38, 43, 48, 53, 58],
"宜人性": [4, 9, 14, 19, 24, 29, 34, 39, 44, 49, 54, 59],
"尽责性": [5, 10, 15, 20, 25, 30, 35, 40, 45, 50, 55, 60],
},
"use_std": False,
},
# === 智商/天赋测试 ===
"duoyuan80": {
"name": "加德纳多元智能测试(MI-80)",
"option_base": 1, # 1-5
"num_options": 5,
"reverse": [],
"transform": None,
"ratings": [], # factor-based, no global rating
"factors": {
"语言智能": [1,2,3,4,5,6,7,8,9,10],
"逻辑数学智能": [11,12,13,14,15,16,17,18,19,20],
"空间智能": [21,22,23,24,25,26,27,28,29,30],
"身体运动智能": [31,32,33,34,35,36,37,38,39,40],
"音乐智能": [41,42,43,44,45,46,47,48,49,50],
"自然观察智能": [51,52,53,54,55,56,57,58,59,60],
"人际智能": [61,62,63,64,65,66,67,68,69,70],
"内省智能": [71,72,73,74,75,76,77,78,79,80],
},
"use_std": False,
},
}
def calculate_score(slug, answers):
"""Calculate scores with known rules, fall back to raw sum."""
scoring = SCORING.get(slug)
num_options = scoring.get("num_options", 4) if scoring else 4
if scoring:
base = scoring["option_base"]
reverse_items = set(scoring["reverse"])
raw = 0
for i, ans in enumerate(answers):
if ans is None:
continue
q_num = i + 1 # 1-indexed question number
opt_idx = int(ans)
if q_num in reverse_items:
# Reverse: last option → first score
raw += base + (num_options - 1 - opt_idx)
else:
raw += base + opt_idx
# Factor scores
factors = {}
if scoring.get("factors"):
for fname, qnums in scoring["factors"].items():
f_raw = 0
f_count = 0
for qn in qnums:
idx = qn - 1
if idx < len(answers) and answers[idx] is not None:
opt_idx = int(answers[idx])
if qn in reverse_items:
f_raw += base + (num_options - 1 - opt_idx)
else:
f_raw += base + opt_idx
f_count += 1
if f_count > 0:
factors[fname] = f_raw
# Transform (e.g. SDS std = raw * 1.25)
std_score = None
if scoring.get("transform"):
mult, rnd = scoring["transform"]
std_score = round(raw * mult) if rnd == 0 else int(raw * mult)
used_score = std_score if (scoring.get("use_std") and std_score is not None) else raw
# Rating
rating = ""
for threshold, label in scoring.get("ratings", []):
if used_score <= threshold:
rating = label
break
if not rating and scoring.get("ratings"):
rating = scoring["ratings"][-1][1]
max_raw = len([a for a in answers if a is not None]) * (base + num_options - 1)
# Max per factor (for front-end bar charts)
factor_max = {}
if scoring.get("factors"):
for fname, qnums in scoring["factors"].items():
factor_max[fname] = len(qnums) * (base + num_options - 1)
# Per-question scores (1-indexed options: 选项A=1分, B=2分, ...)
question_scores = []
for i, ans in enumerate(answers):
if ans is None:
question_scores.append(None)
elif (i + 1) in reverse_items:
question_scores.append(base + (num_options - 1 - int(ans)))
else:
question_scores.append(base + int(ans))
return {
"raw_score": raw,
"std_score": std_score,
"max_score": max_raw,
"rating": rating,
"factors": factors,
"factor_max": factor_max,
"answer_count": len([a for a in answers if a is not None]),
"question_scores": question_scores,
}
else:
# Generic: option index + 1
raw = sum(int(a) + 1 for a in answers if a is not None)
question_scores = [(int(a) + 1) if a is not None else None for a in answers]
return {
"raw_score": raw,
"std_score": None,
"max_score": len([a for a in answers if a is not None]) * num_options,
"rating": "",
"factors": {},
"factor_max": {},
"answer_count": len([a for a in answers if a is not None]),
"question_scores": question_scores,
}
class Handler(SimpleHTTPRequestHandler):
def __init__(self, *args, **kwargs):
super().__init__(*args, directory=ROOT, **kwargs)
def end_headers(self):
no_cache_paths = ['/', '/load', '/api/']
if any(self.path == p or self.path.startswith(p) for p in no_cache_paths):
self.send_header("Cache-Control", "no-cache, no-store, must-revalidate")
self.send_header("Pragma", "no-cache")
self.send_header("Expires", "0")
super().end_headers()
def _json(self, data, code=200):
self.send_response(code)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(json.dumps(data, ensure_ascii=False).encode())
def _read_body(self):
length = int(self.headers.get("Content-Length", 0))
return self.rfile.read(length) if length else b""
# ==== API routing ====
def do_GET(self):
path = urlparse(self.path).path
# Scale API
if path == "/api/scales":
return self.api_scales_list()
if path.startswith("/api/scales/") and not path.startswith("/api/scales/result"):
slug = path.split("/api/scales/")[1]
return self.api_scale_detail(slug)
if path == "/api/scales/results":
return self.api_results_list()
if path.startswith("/api/scales/results/"):
rid = path.split("/api/scales/results/")[1]
return self.api_result_detail(rid)
# Mindmap
if path == "/load":
return self.mindmap_load()
return super().do_GET()
def do_POST(self):
path = urlparse(self.path).path
# Scale result submission
if path == "/api/scales/result":
return self.api_submit_result()
# Mindmap save
if path == "/save":
return self.mindmap_save()
return super().do_POST()
def do_OPTIONS(self):
self.send_response(204)
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type")
self.end_headers()
# ---- Scale APIs ----
def api_scales_list(self):
if not os.path.exists(SCALES_FILE):
return self._json({"error": "No scales data"}, 404)
with open(SCALES_FILE, "r", encoding="utf-8") as f:
scales = json.load(f)
# Return lightweight list (no questions)
result = []
for s in scales:
result.append({
"slug": s["slug"], "name": s["name"], "category": s["category"],
"question_count": s["question_count"], "description": s.get("description", ""),
})
self._json(result)
def api_scale_detail(self, slug):
if not os.path.exists(SCALES_FILE):
return self._json({"error": "No scales data"}, 404)
with open(SCALES_FILE, "r", encoding="utf-8") as f:
scales = json.load(f)
for s in scales:
if s["slug"] == slug:
return self._json(s)
self._json({"error": "Scale not found"}, 404)
def api_submit_result(self):
body = self._read_body()
try:
data = json.loads(body)
except json.JSONDecodeError:
return self._json({"error": "Invalid JSON"}, 400)
slug = data.get("scale_slug", "")
scale_name = data.get("scale_name", "")
answers = data.get("answers", []) # list of option indices
if not slug or not answers:
return self._json({"error": "Missing scale_slug or answers"}, 400)
rid = str(uuid.uuid4())[:8]
created = time.strftime("%Y-%m-%d %H:%M:%S")
# Calculate score using known rules
scores = calculate_score(slug, answers)
# Get the scale config for display name
scale_config = SCORING.get(slug, {})
scores["scale_name"] = scale_name
scores["scale_slug"] = slug
db = get_db()
db.execute(
"INSERT INTO results (id, scale_slug, scale_name, answers, scores, created_at, user_agent, ip) VALUES (?,?,?,?,?,?,?,?)",
(rid, slug, scale_name, json.dumps(answers), json.dumps(scores), created,
self.headers.get("User-Agent", ""), self.client_address[0])
)
db.commit()
db.close()
self._json({"id": rid, "scores": scores, "created_at": created})
def api_results_list(self):
db = get_db()
rows = db.execute("SELECT id, scale_slug, scale_name, scores, created_at FROM results ORDER BY created_at DESC LIMIT 50").fetchall()
db.close()
results = []
for r in rows:
results.append({
"id": r["id"], "scale_slug": r["scale_slug"], "scale_name": r["scale_name"],
"scores": json.loads(r["scores"]), "created_at": r["created_at"],
})
self._json(results)
def api_result_detail(self, rid):
db = get_db()
r = db.execute("SELECT * FROM results WHERE id = ?", (rid,)).fetchone()
db.close()
if not r:
return self._json({"error": "Not found"}, 404)
self._json({
"id": r["id"], "scale_slug": r["scale_slug"], "scale_name": r["scale_name"],
"answers": json.loads(r["answers"]), "scores": json.loads(r["scores"]),
"created_at": r["created_at"],
})
# ---- Mindmap (existing) ----
def mindmap_load(self):
if os.path.exists(MINDMAP_FILE):
with open(MINDMAP_FILE, "r", encoding="utf-8") as f:
data = f.read()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(data.encode())
else:
self._json({})
def mindmap_save(self):
data = self._read_body()
try:
json.loads(data)
except json.JSONDecodeError:
self.send_error(400, "Invalid JSON")
return
with open(MINDMAP_FILE, "w", encoding="utf-8") as f:
f.write(data.decode())
self._json({"ok": True})
def log_message(self, format, *args):
if any(x in str(args) for x in ["/save", "/load", "/api/"]):
print(f"[{self.log_date_time_string()}] {args[0]}")
if __name__ == "__main__":
port = int(sys.argv[1]) if len(sys.argv) > 1 else 8001
server = HTTPServer(("127.0.0.1", port), Handler)
server.socket.settimeout(30)
print(f"愈心谷 server on :{port} (scales API enabled)")
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
Reference in New Issue View Git Blame Copy Permalink