shutiaozhao127-boop
cfdeeb5e41
添加后端API和前端项目的初始目录结构，包括应用层模块、工具模块和模型定义 配置前后端的.dockerignore文件以排除开发环境文件 添加前端样式文件，包括字体、Tailwind CSS和主题配置 初始化Python虚拟环境中的类型标记文件
259 lines
8.7 KiB
Python
259 lines
8.7 KiB
Python
import re
|
||
from typing import Optional, Tuple
|
||
|
||
from beanie import PydanticObjectId
|
||
from fastapi import APIRouter, Depends
|
||
|
||
from app.middleware.auth import get_current_user
|
||
from app.models.report import Report
|
||
from app.models.user import User
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
def _parse_numeric_value(value) -> Optional[float]:
|
||
value_str = str(value).strip() if value is not None else ""
|
||
if not value_str:
|
||
return None
|
||
match = re.search(r'(\d+\.?\d*)', value_str)
|
||
if not match:
|
||
return None
|
||
try:
|
||
return float(match.group(1))
|
||
except (ValueError, AttributeError):
|
||
return None
|
||
|
||
|
||
def _parse_reference_range(reference_range: str) -> Tuple[Optional[float], Optional[float]]:
|
||
if not reference_range:
|
||
return None, None
|
||
match = re.search(r'(\d+\.?\d*)\s*[-~]\s*(\d+\.?\d*)', reference_range)
|
||
if not match:
|
||
return None, None
|
||
return float(match.group(1)), float(match.group(2))
|
||
|
||
|
||
@router.get("/overview")
|
||
async def get_health_overview(user: User = Depends(get_current_user)):
|
||
"""健康档案概览"""
|
||
reports = await Report.find(Report.user_id == user.id).sort("-created_at").to_list()
|
||
total = len(reports)
|
||
normal = sum(1 for r in reports if r.overall_status == "normal")
|
||
attention = sum(1 for r in reports if r.overall_status == "attention")
|
||
warning = sum(1 for r in reports if r.overall_status == "warning")
|
||
|
||
recent = [
|
||
{
|
||
"id": str(r.id),
|
||
"report_type": r.report_type,
|
||
"overall_status": r.overall_status,
|
||
"created_at": r.created_at.isoformat(),
|
||
}
|
||
for r in reports[:5]
|
||
]
|
||
|
||
return {
|
||
"stats": {
|
||
"total": total,
|
||
"normal": normal,
|
||
"attention": attention,
|
||
"warning": warning,
|
||
},
|
||
"recent_reports": recent,
|
||
}
|
||
|
||
|
||
@router.get("/growth-curve")
|
||
async def get_growth_curve(user: User = Depends(get_current_user)):
|
||
"""获取胎儿生长曲线数据（从B超报告提取）"""
|
||
reports = await Report.find(
|
||
Report.user_id == user.id,
|
||
Report.report_type == "B超",
|
||
).sort("pregnancy_week").to_list()
|
||
|
||
data_points = []
|
||
for r in reports:
|
||
if not r.pregnancy_week:
|
||
continue
|
||
point = {"week": r.pregnancy_week}
|
||
for item in r.data_items:
|
||
name_lower = item.name.lower()
|
||
numeric_value = _parse_numeric_value(item.value)
|
||
if numeric_value is None:
|
||
continue
|
||
|
||
# 匹配指标名称
|
||
if "双顶径" in item.name or "bpd" in name_lower:
|
||
point["bpd"] = numeric_value
|
||
elif "腹围" in item.name or "ac" in name_lower:
|
||
point["ac"] = numeric_value
|
||
elif "股骨长" in item.name or "fl" in name_lower:
|
||
point["fl"] = numeric_value
|
||
|
||
# 只有包含至少一个生长指标的数据点才添加
|
||
if any(k in point for k in ["bpd", "ac", "fl"]):
|
||
data_points.append(point)
|
||
|
||
return {"data_points": data_points}
|
||
|
||
|
||
@router.get("/timeline")
|
||
async def get_timeline(user: User = Depends(get_current_user)):
|
||
"""报告时间轴"""
|
||
reports = await Report.find(Report.user_id == user.id).sort("-created_at").to_list()
|
||
|
||
timeline = []
|
||
for r in reports:
|
||
week = r.pregnancy_week or 0
|
||
if week <= 12:
|
||
period = "early"
|
||
elif week <= 28:
|
||
period = "mid"
|
||
else:
|
||
period = "late"
|
||
|
||
timeline.append({
|
||
"id": str(r.id),
|
||
"report_type": r.report_type,
|
||
"date": r.created_at.isoformat(),
|
||
"status": r.overall_status,
|
||
"summary": f"{len(r.data_items)}项指标",
|
||
"period": period,
|
||
"week": week,
|
||
})
|
||
|
||
return {"reports": timeline}
|
||
|
||
|
||
@router.get("/indicator-trends")
|
||
async def get_indicator_trends(user: User = Depends(get_current_user)):
|
||
"""获取跨报告指标趋势数据（同名数值指标出现2+次的）"""
|
||
reports = await Report.find(
|
||
Report.user_id == user.id,
|
||
).sort("pregnancy_week", "created_at").to_list()
|
||
|
||
# 收集所有指标：indicator_name -> list of {week, value, unit, ref_range, status, date, report_type}
|
||
indicator_map: dict[str, list[dict]] = {}
|
||
|
||
for r in reports:
|
||
week = r.pregnancy_week or 0
|
||
date_str = r.created_at.isoformat() if r.created_at else ""
|
||
for item in r.data_items:
|
||
value_str = str(item.value).strip() if item.value else ""
|
||
if not value_str:
|
||
continue
|
||
match = re.search(r'(\d+\.?\d*)', value_str)
|
||
if not match:
|
||
continue
|
||
try:
|
||
numeric_value = float(match.group(1))
|
||
except (ValueError, AttributeError):
|
||
continue
|
||
|
||
name = item.name.strip()
|
||
if name not in indicator_map:
|
||
indicator_map[name] = []
|
||
indicator_map[name].append({
|
||
"week": week,
|
||
"value": numeric_value,
|
||
"unit": item.unit or "",
|
||
"reference_range": item.reference_range or "",
|
||
"status": item.status or "normal",
|
||
"date": date_str,
|
||
"report_type": r.report_type,
|
||
})
|
||
|
||
# 保留所有有数据的指标（含仅1次的，用于综合分析）
|
||
indicators = []
|
||
for name, points in indicator_map.items():
|
||
|
||
# 解析参考范围（取第一个非空的）
|
||
ref_lower = None
|
||
ref_upper = None
|
||
for p in points:
|
||
if p["reference_range"]:
|
||
# 尝试解析 "3.5-5.0" 或 "3.5~5.0" 格式
|
||
ref_match = re.search(r'(\d+\.?\d*)\s*[-~]\s*(\d+\.?\d*)', p["reference_range"])
|
||
if ref_match:
|
||
ref_lower = float(ref_match.group(1))
|
||
ref_upper = float(ref_match.group(2))
|
||
break
|
||
|
||
indicators.append({
|
||
"name": name,
|
||
"unit": points[0]["unit"],
|
||
"ref_lower": ref_lower,
|
||
"ref_upper": ref_upper,
|
||
"points": [{"week": p["week"], "value": p["value"], "date": p["date"],
|
||
"status": p["status"], "report_type": p["report_type"]} for p in points],
|
||
})
|
||
|
||
return {"indicators": indicators}
|
||
|
||
|
||
@router.get("/report-type-trends")
|
||
async def get_report_type_trends(user: User = Depends(get_current_user)):
|
||
"""按报告类型分组的指标趋势（同类型报告中出现2+次的数值指标）"""
|
||
reports = await Report.find(
|
||
Report.user_id == user.id,
|
||
).sort("pregnancy_week", "created_at").to_list()
|
||
|
||
# report_type -> indicator_name -> list of data points
|
||
type_map: dict[str, dict[str, list[dict]]] = {}
|
||
|
||
for r in reports:
|
||
rt = r.report_type or "其他"
|
||
week = r.pregnancy_week or 0
|
||
date_str = r.created_at.isoformat() if r.created_at else ""
|
||
|
||
if rt not in type_map:
|
||
type_map[rt] = {}
|
||
|
||
for item in r.data_items:
|
||
numeric_value = _parse_numeric_value(item.value)
|
||
if numeric_value is None:
|
||
continue
|
||
|
||
name = item.name.strip()
|
||
if name not in type_map[rt]:
|
||
type_map[rt][name] = []
|
||
type_map[rt][name].append({
|
||
"week": week,
|
||
"value": numeric_value,
|
||
"unit": item.unit or "",
|
||
"reference_range": item.reference_range or "",
|
||
"status": item.status or "normal",
|
||
"date": date_str,
|
||
})
|
||
|
||
# 组装结果：保留所有有数据的指标（含仅1个数据点的）
|
||
result = []
|
||
for rt, ind_map in type_map.items():
|
||
indicators = []
|
||
for name, points in ind_map.items():
|
||
ref_lower, ref_upper = None, None
|
||
for p in points:
|
||
ref_lower, ref_upper = _parse_reference_range(p["reference_range"])
|
||
if ref_lower is not None:
|
||
break
|
||
indicators.append({
|
||
"name": name,
|
||
"unit": points[0]["unit"],
|
||
"ref_lower": ref_lower,
|
||
"ref_upper": ref_upper,
|
||
"points": [{"week": p["week"], "value": p["value"], "date": p["date"],
|
||
"status": p["status"]} for p in points],
|
||
})
|
||
if indicators:
|
||
# 按数据点数量降序，让最有数据的指标排前面
|
||
indicators.sort(key=lambda x: len(x["points"]), reverse=True)
|
||
result.append({
|
||
"report_type": rt,
|
||
"report_count": sum(1 for r in reports if (r.report_type or "其他") == rt),
|
||
"indicators": indicators,
|
||
})
|
||
|
||
# 按报告份数降序
|
||
result.sort(key=lambda x: x["report_count"], reverse=True)
|
||
return {"report_types": result}
|