fa-agent/finance/views.py

import csv
import hashlib
import io
import ipaddress
import json
import math
import tempfile
import uuid
from datetime import timedelta
from functools import wraps
from pathlib import Path

from django.conf import settings
from django.contrib.auth import authenticate, login, logout
from django.contrib.auth.models import User
from django.contrib.auth.password_validation import validate_password
from django.core.exceptions import ValidationError
from django.db import IntegrityError, connection, transaction
from django.http import FileResponse, HttpResponse, JsonResponse
from django.middleware.csrf import get_token
from django.utils import timezone
from openpyxl import Workbook
from openpyxl.cell import WriteOnlyCell

from .analysis import METRICS, analyze, context, default_preferences, dimensions, scoped_facts
from .importing import file_digest, profile_file, suggest_mapping, validate_mapping
from .models import AIConfig, Audit, Dataset, Investigation, Membership, Preference, Throttle


class APIError(Exception):
    def __init__(self, message, status=400):
        self.message, self.status = message, status


def endpoint(methods=("GET",), roles=None, public=False):
    def decorate(fn):
        @wraps(fn)
        def view(request, *args, **kwargs):
            try:
                if request.method not in methods:
                    raise APIError("请求方法不允许", 405)
                if not public:
                    if not request.user.is_authenticated:
                        raise APIError("请先登录", 401)
                    try:
                        request.member = request.user.membership
                    except Membership.DoesNotExist:
                        raise APIError("账号尚未分配组织", 403)
                    if roles and request.member.role not in roles:
                        raise APIError("没有此操作的权限", 403)
                return fn(request, *args, **kwargs)
            except APIError as exc:
                return JsonResponse({"error": exc.message}, status=exc.status)
            except (ValueError, ValidationError, json.JSONDecodeError) as exc:
                return JsonResponse({"error": str(exc)}, status=400)

        return view

    return decorate


def body(request):
    if request.content_type != "application/json":
        raise APIError("请求必须使用 application/json")
    value = json.loads(request.body or b"{}")
    if not isinstance(value, dict):
        raise APIError("请求必须是 JSON 对象")
    return value


def audit(request, action, resource="", details=None):
    Audit.objects.create(
        organization=request.member.organization,
        user=request.user,
        action=action,
        resource=str(resource),
        details=details or {},
    )


def limit(key, count=10, seconds=900):
    key = hashlib.sha256(key.encode()).hexdigest()
    now = timezone.now()
    with transaction.atomic():
        item, _ = Throttle.objects.get_or_create(
            key=key, defaults={"expires_at": now + timedelta(seconds=seconds)}
        )
        item = Throttle.objects.select_for_update().get(pk=key)
        if item.expires_at <= now:
            item.count, item.expires_at = 0, now + timedelta(seconds=seconds)
        if item.count >= count:
            raise APIError("请求过于频繁，请稍后重试", 429)
        item.count += 1
        item.save()


def dataset(request, dataset_id, ready=False):
    ds = Dataset.objects.filter(pk=dataset_id, organization=request.member.organization).first()
    if not ds:
        raise APIError("数据集不存在", 404)
    if ready and ds.status != "ready":
        raise APIError("数据集尚未导入完成", 409)
    return ds


def preferences(request, ds):
    pref = Preference.objects.filter(dataset=ds, user=request.user).first()
    return {**default_preferences(ds), **(pref.value if pref else {})}, pref.revision if pref else 0


def serialize_dataset(ds, detailed=False, restricted=False, raw_access=True):
    out = {
        "id": str(ds.pk),
        "name": ds.name,
        "status": ds.status,
        "version": ds.sha256,
        "row_count": None if restricted else ds.row_count,
        "periods": ds.periods,
        "created_at": ds.created_at.isoformat(),
        "error": ds.error if not restricted else "",
        "quality": ds.quality if not restricted else {},
        "dimensions": dimensions(ds) if ds.mapping else [],
    }
    if detailed:
        out.update(mapping=ds.mapping)
        if not restricted and raw_access:
            out["profile"] = ds.profile
    return out


@endpoint(public=True)
def session(request):
    ai_enabled = settings.AI_ENABLED and bool(settings.OPENAI_API_KEY and settings.OPENAI_MODEL)
    if request.user.is_authenticated:
        try:
            cfg = AIConfig.objects.filter(organization=request.user.membership.organization).first()
            if cfg:
                ai_enabled = (
                    cfg.enabled and bool(cfg.model) and (cfg.provider == "local" or bool(cfg.api_key))
                )
        except Membership.DoesNotExist:
            pass
    data = {
        "csrf": get_token(request),
        "user": None,
        "ai_enabled": ai_enabled,
    }
    if request.user.is_authenticated:
        try:
            m = request.user.membership
            data["user"] = {
                "id": request.user.pk,
                "name": request.user.username,
                "role": m.role,
                "organization": m.organization.name,
                "departments": m.departments,
            }
        except Membership.DoesNotExist:
            pass
    return JsonResponse(data)


@endpoint(("POST",), public=True)
def sign_in(request):
    data = body(request)
    username, password = data.get("username", ""), data.get("password", "")
    if (
        not isinstance(username, str)
        or not isinstance(password, str)
        or len(username) > 150
        or len(password) > 1024
    ):
        raise APIError("登录参数无效")
    address = request.META.get("REMOTE_ADDR", "unknown")
    if settings.TRUST_PROXY and request.META.get("HTTP_X_REAL_IP"):
        try:
            address = str(ipaddress.ip_address(request.META["HTTP_X_REAL_IP"]))
        except ValueError:
            raise APIError("代理来源地址无效")
    limit("login-ip:" + address, 30)
    limit("login-user:" + username.casefold(), 10)
    user = authenticate(request, username=username, password=password)
    if not user or not Membership.objects.filter(user=user).exists():
        raise APIError("用户名或密码错误", 401)
    login(request, user)
    request.member = user.membership
    audit(request, "auth.login")
    return JsonResponse({"ok": True, "csrf": get_token(request)})


@endpoint(("POST",))
def sign_out(request):
    audit(request, "auth.logout")
    logout(request)
    return JsonResponse({"ok": True})


@endpoint(("GET", "POST"))
def datasets(request):
    if request.method == "GET":
        return JsonResponse(
            {
                "items": [
                    serialize_dataset(ds, restricted=bool(request.member.departments))
                    for ds in Dataset.objects.filter(organization=request.member.organization).order_by(
                        "-created_at"
                    )[:100]
                ]
            }
        )
    if request.member.role not in ("admin", "analyst") or request.member.departments:
        raise APIError("仅有完整数据权限的管理员或分析师可导入", 403)
    limit(f"upload:{request.user.pk}", 20, 3600)
    file = request.FILES.get("file")
    if not file or not 0 < file.size <= settings.MAX_UPLOAD_BYTES:
        raise APIError("文件为空或超过上传大小限制")
    ext = Path(file.name).suffix.lower()
    if ext not in (".xlsx", ".csv"):
        raise APIError("仅接受 .xlsx 或 UTF-8 CSV")
    digest = file_digest(file)
    existing = Dataset.objects.filter(organization=request.member.organization, sha256=digest).first()
    if existing:
        return JsonResponse(
            serialize_dataset(
                existing,
                True,
                raw_access=request.member.role == "admin"
                or (existing.created_by_id == request.user.pk and existing.status in ("mapping", "failed")),
            ),
            status=200,
        )
    ds = Dataset(
        organization=request.member.organization,
        created_by=request.user,
        name=Path(file.name).name[:200],
        sha256=digest,
    )
    ds.file.save(str(uuid.uuid4()) + ext, file, save=False)
    try:
        ds.profile = profile_file(ds.file.path)
        ds.mapping = suggest_mapping(ds.profile)
        ds.save()
    except IntegrityError:
        ds.file.delete(save=False)
        return JsonResponse(
            serialize_dataset(
                Dataset.objects.get(organization=request.member.organization, sha256=digest),
                True,
                raw_access=request.member.role == "admin",
            )
        )
    except Exception as exc:
        ds.file.delete(save=False)
        raise APIError(
            "无法读取文件，请检查格式或文件安全限制："
            + (str(exc)[:160] if isinstance(exc, (ValueError, UnicodeError)) else "文件损坏")
        ) from exc
    audit(request, "dataset.upload", ds.pk, {"sha256": digest})
    return JsonResponse(serialize_dataset(ds, True), status=201)


@endpoint()
def dataset_detail(request, dataset_id):
    ds = dataset(request, dataset_id)
    raw_access = request.member.role == "admin" or (
        ds.created_by_id == request.user.pk and ds.status in ("mapping", "failed")
    )
    result = serialize_dataset(ds, True, bool(request.member.departments), raw_access)
    if ds.status == "ready":
        result["preferences"], result["revision"] = preferences(request, ds)
    return JsonResponse(result)


@endpoint(("POST",), roles=("admin", "analyst"))
def mapping(request, dataset_id):
    ds = dataset(request, dataset_id)
    if request.member.departments:
        raise APIError("没有修改导入映射的权限", 403)
    data = body(request)
    if data.get("suggest"):
        return JsonResponse(suggest_mapping(ds.profile, data.get("sheet"), data.get("header")))
    value = validate_mapping(data, ds.profile)
    updated = Dataset.objects.filter(pk=ds.pk, status__in=["mapping", "failed"]).update(
        mapping=value, status="queued", error="", started_at=None, finished_at=None
    )
    if not updated:
        raise APIError("数据版本已锁定，不能重复导入或修改", 409)
    audit(request, "dataset.confirm_mapping", ds.pk, {"mapping": value})
    return JsonResponse({"status": "queued"}, status=202)


@endpoint(("POST",))
def query(request, dataset_id):
    ds = dataset(request, dataset_id, True)
    pref, _ = preferences(request, ds)
    limit(f"analysis:{request.user.pk}", 120, 60)
    return JsonResponse(analyze(ds, request.member, body(request), pref))


def amount_text(value):
    """金额展示文本：去掉定点数存储尺度产生的多余尾零（如 10.000000000000000000000000 → 10）。"""
    if value is None:
        return None
    text = format(value.normalize(), "f")
    return "0" if text == "-0" else text


@endpoint(("POST",))
def evidence(request, dataset_id):
    ds = dataset(request, dataset_id, True)
    data = body(request)
    ctx = context(ds, data)
    offset, size = data.get("offset", 0), data.get("limit", 50)
    if type(offset) is not int or type(size) is not int or not 0 <= offset <= 1000000 or not 1 <= size <= 200:
        raise APIError("分页参数无效")
    side = data.get("side", "current")
    if side not in ("current", "baseline"):
        raise APIError("期间选择无效")
    qs = scoped_facts(ds, request.member, ctx["filters"]).filter(period__in=ctx[side]).order_by("source_row")
    rows = list(
        qs[offset : offset + size].values(
            "source_row", "period", "dimensions", "revenue", "cost", "gross", "raw"
        )
    )
    for row in rows:
        for field in ("revenue", "cost", "gross"):
            row[field] = amount_text(row[field])
    # Raw patient/customer identifiers require an explicit administrator role.
    if request.member.role != "admin":
        for row in rows:
            row.pop("raw")
    audit(
        request,
        "evidence.read",
        ds.pk,
        {"context": ctx, "side": side, "offset": offset, "limit": size},
    )
    return JsonResponse(
        {
            "rows": rows,
            "total": qs.count(),
            "context": ctx,
            "version": ds.sha256,
            "sheet": ds.mapping["sheet"],
        }
    )


def csv_safe(value):
    text = str(value) if value is not None else ""
    if not isinstance(value, str):
        return text
    return (
        "'" + text
        if text.startswith(("\t", "\r", "\n")) or text.lstrip().startswith(("=", "+", "-", "@"))
        else text
    )


@endpoint(("POST",), roles=("admin", "analyst"))
def export(request, dataset_id):
    ds = dataset(request, dataset_id, True)
    data = body(request)
    ctx = context(ds, data)
    limit(f"export:{request.user.pk}", 10, 3600)
    kind = data.get("format", "csv")
    if kind not in ("csv", "xlsx"):
        raise APIError("导出格式无效")
    qs = (
        scoped_facts(ds, request.member, ctx["filters"])
        .filter(period__in=ctx["current"] + ctx["baseline"])
        .order_by("source_row")
    )
    if qs.count() > 100000:
        raise APIError("单次导出上限 100,000 行，请缩小期间或筛选范围")
    output = tempfile.TemporaryFile(mode="w+b")
    csv_stream = None
    if kind == "csv":
        output.write(b"\xef\xbb\xbf")
        csv_stream = io.TextIOWrapper(output, encoding="utf-8", newline="", write_through=True)
    csv_writer = csv.writer(csv_stream) if csv_stream else None
    workbook = Workbook(write_only=True) if kind == "xlsx" else None
    sheet = workbook.create_sheet("证据明细") if workbook else None

    def write_row(values):
        if sheet is None:
            csv_writer.writerow(csv_safe(value) for value in values)
        else:
            cells = []
            for value in values:
                # Excel has only 15 significant digits; evidence amounts stay exact text.
                cell = WriteOnlyCell(sheet, value=str(value) if value is not None else "")
                cell.data_type = "s"
                cells.append(cell)
            sheet.append(cells)

    try:
        write_row(
            [
                "TraceIQ 数据版本",
                ds.sha256,
            ]
        )
        write_row(["筛选上下文", json.dumps(ctx, ensure_ascii=False)])
        write_row(["ROI 口径", ds.mapping["roi_formula"]])
        write_row(["导出时间 UTC", timezone.now().isoformat()])
        dims = dimensions(ds)
        write_row(
            [
                "原始工作表",
                "原始行号",
                "期间",
                *[d["name"] for d in dims],
                "实际收入",
                "成本",
                "毛利贡献",
            ]
        )
        for row in qs.iterator(chunk_size=1000):
            write_row(
                [
                    ds.mapping["sheet"],
                    row.source_row,
                    row.period,
                    *[row.dimensions.get(d["key"], "") for d in dims],
                    amount_text(row.revenue),
                    amount_text(row.cost),
                    amount_text(row.gross),
                ]
            )
        if workbook:
            workbook.save(output)
            content_type = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
        else:
            csv_stream.flush()
            csv_stream.detach()
            csv_stream = None
            content_type = "text/csv; charset=utf-8"
        output.seek(0)
        audit(request, "evidence.export", ds.pk, {"context": ctx, "format": kind})
        return FileResponse(
            output,
            as_attachment=True,
            filename=f"traceiq-evidence.{kind}",
            content_type=content_type,
        )
    except Exception:
        if csv_stream is not None:
            csv_stream.close()
        else:
            output.close()
        raise


@endpoint(("GET", "PUT"))
def settings_view(request, dataset_id):
    ds = dataset(request, dataset_id, True)
    if request.method == "GET":
        value, revision = preferences(request, ds)
        return JsonResponse({"value": value, "revision": revision})
    data = body(request)
    value = data.get("value")
    defaults = default_preferences(ds)
    if not isinstance(value, dict) or set(value) != set(defaults):
        raise APIError("设置字段不完整")
    allowed = {d["key"] for d in dimensions(ds)}
    hierarchy = value["hierarchy"]
    if (
        not isinstance(hierarchy, list)
        or not hierarchy
        or any(not isinstance(k, str) or k not in allowed for k in hierarchy)
        or len(set(hierarchy)) != len(hierarchy)
    ):
        raise APIError("下钻层级无效或重复")
    if (
        value["entry"] not in ("overview", "anomalies")
        or value["comparison"] not in ("yoy", "mom")
        or not isinstance(value["metric"], str)
        or value["metric"] not in METRICS
    ):
        raise APIError("默认视图或指标无效")
    for field, lo, hi in [("threshold", 0, 10), ("min_impact", 0, 1e15), ("min_samples", 1, 1000000)]:
        v = value[field]
        if type(v) not in (int, float) or not math.isfinite(v) or not lo <= v <= hi:
            raise APIError("异常阈值无效")
    if type(value["min_samples"]) is not int:
        raise APIError("最小样本数必须为整数")
    for field in ["show_heatmap", "show_queue", "show_copilot", "alerts"]:
        if type(value[field]) is not bool:
            raise APIError("显示设置必须为布尔值")
    with transaction.atomic():
        pref = Preference.objects.select_for_update().filter(dataset=ds, user=request.user).first()
        if data.get("revision") != (pref.revision if pref else 0):
            raise APIError("设置已在其他窗口更新，请重新打开", 409)
        if pref:
            pref.value, pref.revision = value, pref.revision + 1
            pref.save()
        else:
            try:
                pref = Preference.objects.create(dataset=ds, user=request.user, value=value)
            except IntegrityError as exc:
                raise APIError("设置发生并发更新，请重新打开", 409) from exc
        audit(request, "settings.update", ds.pk, {"revision": pref.revision, "value": value})
    return JsonResponse({"value": pref.value, "revision": pref.revision})


@endpoint(("GET", "POST"))
def investigations(request, dataset_id):
    ds = dataset(request, dataset_id, True)
    if request.method == "GET":
        return JsonResponse(
            {
                "items": list(
                    Investigation.objects.filter(dataset=ds, user=request.user).values(
                        "id", "key", "context", "status", "note", "updated_at"
                    )
                )
            }
        )
    if request.member.role == "viewer":
        raise APIError("查看者不可修改调查状态", 403)
    data = body(request)
    ctx = context(ds, data.get("context", {}))
    status, note = data.get("status"), data.get("note", "")
    if status not in ("open", "reviewing", "resolved") or not isinstance(note, str) or len(note) > 2000:
        raise APIError("调查状态或备注无效")
    key = hashlib.sha256(json.dumps(ctx, sort_keys=True).encode()).hexdigest()
    item, _ = Investigation.objects.update_or_create(
        dataset=ds, user=request.user, key=key, defaults={"context": ctx, "status": status, "note": note}
    )
    audit(request, "investigation.update", ds.pk, {"key": key, "status": status})
    return JsonResponse({"id": item.pk, "key": key, "status": status})


@endpoint(("GET", "POST", "PATCH"), roles=("admin",))
def users(request):
    if request.member.departments:
        raise APIError("限域管理员不能管理账号", 403)
    if request.method == "GET":
        return JsonResponse(
            {
                "items": [
                    {
                        "id": m.user_id,
                        "username": m.user.username,
                        "role": m.role,
                        "active": m.user.is_active,
                        "departments": m.departments,
                    }
                    for m in Membership.objects.filter(
                        organization=request.member.organization
                    ).select_related("user")
                ]
            }
        )
    data = body(request)
    role, departments = data.get("role"), data.get("departments", [])
    if (
        role not in ("admin", "analyst", "viewer")
        or not isinstance(departments, list)
        or any(not isinstance(d, str) or not d or len(d) > 256 for d in departments)
        or len(departments) > 100
    ):
        raise APIError("角色或科室范围无效")
    with transaction.atomic():
        if request.method == "POST":
            username, password = data.get("username"), data.get("password")
            if (
                not isinstance(username, str)
                or not 1 <= len(username) <= 150
                or not isinstance(password, str)
            ):
                raise APIError("账号格式无效")
            validate_password(password, User(username=username))
            try:
                user = User.objects.create_user(username=username, password=password)
            except IntegrityError as exc:
                raise APIError("用户名已存在", 409) from exc
            Membership.objects.create(
                user=user, organization=request.member.organization, role=role, departments=departments
            )
        else:
            if type(data.get("id")) is not int or data["id"] <= 0:
                raise APIError("账号 ID 无效")
            membership = (
                Membership.objects.filter(user_id=data.get("id"), organization=request.member.organization)
                .select_related("user")
                .first()
            )
            if not membership:
                raise APIError("账号不存在", 404)
            if membership.user_id == request.user.pk:
                raise APIError("不可在此修改自己的权限")
            if type(data.get("active")) is not bool:
                raise APIError("账号状态无效")
            user = membership.user
            user.is_active = data["active"]
            user.save(update_fields=["is_active"])
            membership.role, membership.departments = role, departments
            membership.save()
        audit(request, "user.update", user.pk, {"role": role, "departments": departments})
    return JsonResponse({"ok": True})


@endpoint(roles=("admin",))
def audit_log(request):
    if request.member.departments:
        raise APIError("没有组织审计权限", 403)
    return JsonResponse(
        {
            "items": list(
                Audit.objects.filter(organization=request.member.organization)
                .order_by("-id")
                .values("id", "user__username", "action", "resource", "details", "created_at")[:200]
            )
        }
    )


@endpoint(public=True)
def health(request):
    with connection.cursor() as cursor:
        cursor.execute("SELECT 1")
    return JsonResponse({"status": "ok"})


def index(request):
    path = settings.BASE_DIR / "frontend/dist/index.html"
    if not path.exists():
        return HttpResponse("Frontend not built. Run npm run build.", status=503)
    return FileResponse(path.open("rb"), content_type="text/html")