379 lines
15 KiB
Python
379 lines
15 KiB
Python
import csv
|
|
import hashlib
|
|
import re
|
|
import zipfile
|
|
from datetime import date, datetime
|
|
from decimal import Decimal, InvalidOperation, getcontext
|
|
from itertools import islice
|
|
|
|
from django.conf import settings
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
from openpyxl import load_workbook
|
|
|
|
from .formula_columns import RESERVED_DATA_HEADERS
|
|
from .models import Dataset, Fact
|
|
|
|
getcontext().prec = 60
|
|
|
|
ROLES = {
|
|
"period",
|
|
"year",
|
|
"month",
|
|
"revenue",
|
|
"cost",
|
|
"gross",
|
|
"quantity",
|
|
"customer",
|
|
"list_amount",
|
|
"discount",
|
|
"dimension",
|
|
"ignore",
|
|
}
|
|
ALIASES = {
|
|
"period": ["期间", "日期", "period", "date"],
|
|
"year": ["年份", "年", "year"],
|
|
"month": ["月份", "月", "month"],
|
|
"revenue": ["实际收入", "收入", "revenue", "actual_revenue"],
|
|
"cost": ["成本", "cost"],
|
|
"gross": ["毛利", "毛利贡献", "gross_profit"],
|
|
"quantity": ["数量", "quantity"],
|
|
"customer": ["客户姓名", "客户编码", "customer_id"],
|
|
"list_amount": ["应收金额", "list_amount"],
|
|
"discount": ["折扣金额（去税）", "discount"],
|
|
}
|
|
DIMENSIONS = {
|
|
"科室（重分类）": "department",
|
|
"科室": "department",
|
|
"department": "department",
|
|
"学科分类": "subject",
|
|
"学科": "subject",
|
|
"医生等级": "doctor_level",
|
|
"医生": "doctor",
|
|
"医生ID": "doctor",
|
|
"产品一级分类": "product",
|
|
"产品二级分类": "product_subcategory",
|
|
"区域": "region",
|
|
"顾客来源": "source",
|
|
}
|
|
|
|
|
|
def safe_text(value):
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, (datetime, date)):
|
|
return value.isoformat()
|
|
return str(value)[:4000]
|
|
|
|
|
|
def validate_archive(file):
|
|
file.seek(0)
|
|
with zipfile.ZipFile(file) as archive:
|
|
infos = archive.infolist()
|
|
if len(infos) > 5000 or sum(i.file_size for i in infos) > 1024 * 1024 * 1024:
|
|
raise ValueError("工作簿解压体积超过安全限制")
|
|
if any(i.file_size / max(i.compress_size, 1) > 1000 for i in infos):
|
|
raise ValueError("工作簿压缩比异常")
|
|
if any("vbaProject" in i.filename or "externalLinks/" in i.filename for i in infos):
|
|
raise ValueError("不接受宏或外部链接工作簿")
|
|
file.seek(0)
|
|
|
|
|
|
def rows_for(path, sheet=None, formulas=False):
|
|
if str(path).lower().endswith(".csv"):
|
|
with open(path, encoding="utf-8-sig", newline="") as file:
|
|
yield from csv.reader(file)
|
|
else:
|
|
wb = load_workbook(path, read_only=True, data_only=not formulas, keep_links=False)
|
|
try:
|
|
yield from wb[sheet].iter_rows(values_only=True)
|
|
finally:
|
|
wb.close()
|
|
|
|
|
|
def profile_file(path):
|
|
if str(path).lower().endswith(".csv"):
|
|
sheets = [{"name": "CSV", "rows": None, "columns": None}]
|
|
else:
|
|
with open(path, "rb") as file:
|
|
validate_archive(file)
|
|
wb = load_workbook(path, read_only=True, data_only=True, keep_links=False)
|
|
sheets = [{"name": s.title, "rows": s.max_row, "columns": s.max_column} for s in wb]
|
|
wb.close()
|
|
previews = []
|
|
for sheet in sheets:
|
|
preview_count = 600 if sheet["name"] == "产品毛利率" else 8
|
|
rows = [list(map(safe_text, r)) for r in islice(rows_for(path, sheet["name"]), preview_count)]
|
|
if any(len(r) > 200 for r in rows):
|
|
raise ValueError("最多支持 200 列")
|
|
previews.append({**sheet, "preview": rows})
|
|
return {"sheets": previews}
|
|
|
|
|
|
def suggest_mapping(profile, sheet=None, header=None):
|
|
s = next((s for s in profile["sheets"] if s["name"] == sheet), None)
|
|
s = s or next((s for s in profile["sheets"] if s["name"] == "明细"), profile["sheets"][0])
|
|
rows = s["preview"]
|
|
if not rows:
|
|
raise ValueError("工作表为空")
|
|
if header is None:
|
|
|
|
def score(row):
|
|
names = set(row)
|
|
return sum(any(a in names for a in aliases) for aliases in ALIASES.values())
|
|
|
|
header = max(range(min(5, len(rows))), key=lambda i: score(rows[i])) + 1
|
|
if not isinstance(header, int) or not 1 <= header <= len(rows):
|
|
raise ValueError("表头行超出预览范围")
|
|
columns, used = [], set()
|
|
for i, name in enumerate(rows[header - 1]):
|
|
role = next(
|
|
(r for r, names in ALIASES.items() if name in names),
|
|
"ignore" if name in {"单价", "折扣金额", "实收金额"} else "dimension",
|
|
)
|
|
if role != "dimension" and role in used:
|
|
role = "ignore"
|
|
# The reference workbook's L column is the agreed list-amount source.
|
|
if name == "应收金额" and i == 11:
|
|
for c in columns:
|
|
if c["role"] == "list_amount":
|
|
c["role"] = "ignore"
|
|
role = "list_amount"
|
|
used.add(role)
|
|
columns.append(
|
|
{
|
|
"index": i,
|
|
"name": name or f"列{i + 1}",
|
|
"description": rows[header - 2][i] if header > 1 and i < len(rows[header - 2]) else "",
|
|
"role": role,
|
|
"key": DIMENSIONS.get(name, f"field_{i}"),
|
|
"confidence": 1 if name in DIMENSIONS or role not in ("dimension", "ignore") else 0.5,
|
|
}
|
|
)
|
|
rates = {}
|
|
for lookup in profile["sheets"]:
|
|
if lookup["name"] != "产品毛利率":
|
|
continue
|
|
name_index = rate_index = None
|
|
for row in lookup["preview"]:
|
|
if "产品分类" in row and "毛利率" in row:
|
|
name_index, rate_index = row.index("产品分类"), row.index("毛利率")
|
|
elif name_index is not None and len(row) > max(name_index, rate_index) and row[name_index]:
|
|
try:
|
|
rate = decimal(row[rate_index])
|
|
if rate is not None and 0 <= rate <= 1:
|
|
rates[row[name_index]] = str(rate)
|
|
except ValueError:
|
|
pass
|
|
return {
|
|
"sheet": s["name"],
|
|
"header": header,
|
|
"columns": columns,
|
|
"margin_rates": rates,
|
|
"self_operated_products": [p for p in ("保健品", "检验检查", "医疗服务", "中药") if p in rates],
|
|
"roi_formula": "profit_over_cost",
|
|
"description": "",
|
|
}
|
|
|
|
|
|
def validate_mapping(mapping, profile):
|
|
if not isinstance(mapping, dict):
|
|
raise ValueError("映射格式错误")
|
|
sheets = {s["name"]: s for s in profile["sheets"]}
|
|
if (
|
|
not isinstance(mapping.get("sheet"), str)
|
|
or mapping["sheet"] not in sheets
|
|
or type(mapping.get("header")) is not int
|
|
or not 1 <= mapping["header"] <= 8
|
|
):
|
|
raise ValueError("工作表或表头无效")
|
|
columns = mapping.get("columns", [])
|
|
if not isinstance(columns, list) or not 1 <= len(columns) <= 200:
|
|
raise ValueError("映射列数无效")
|
|
roles, keys, indexes = [], [], []
|
|
for c in columns:
|
|
if (
|
|
not isinstance(c, dict)
|
|
or not isinstance(c.get("role"), str)
|
|
or c.get("role") not in ROLES
|
|
or type(c.get("index")) is not int
|
|
or not 0 <= c["index"] < 200
|
|
or not isinstance(c.get("name"), str)
|
|
or not 1 <= len(c["name"]) <= 256
|
|
or not isinstance(c.get("description", ""), str)
|
|
or len(c.get("description", "")) > 4000
|
|
):
|
|
raise ValueError("字段映射无效")
|
|
indexes.append(c["index"])
|
|
if c["role"] not in ("ignore", "dimension"):
|
|
roles.append(c["role"])
|
|
if c["role"] == "dimension":
|
|
key = c.get("key", "")
|
|
if not isinstance(key, str) or not re.fullmatch(r"[a-z][a-z0-9_]{0,49}", key) or "__" in key:
|
|
raise ValueError("维度标识只允许小写字母、数字和单下划线")
|
|
keys.append(key)
|
|
if len(set(roles)) != len(roles) or len(set(keys)) != len(keys) or len(set(indexes)) != len(indexes):
|
|
raise ValueError("指标、维度标识或列索引重复")
|
|
if RESERVED_DATA_HEADERS.intersection(key.casefold() for key in keys):
|
|
raise ValueError("维度标识与公式工作簿保留列名冲突")
|
|
if "revenue" not in roles or not ("period" in roles or {"year", "month"}.issubset(roles)):
|
|
raise ValueError("必须映射实际收入以及期间（或年份和月份）")
|
|
if not keys:
|
|
raise ValueError("至少保留一个分析维度")
|
|
rates = mapping.get("margin_rates", {})
|
|
if not isinstance(rates, dict) or len(rates) > 500:
|
|
raise ValueError("毛利率映射无效")
|
|
for key, value in rates.items():
|
|
if not isinstance(key, str) or not 1 <= len(key) <= 256:
|
|
raise ValueError("产品分类名称无效")
|
|
rate = decimal(value)
|
|
if rate is None or not Decimal(0) <= rate <= Decimal(1):
|
|
raise ValueError("毛利率必须在 0 到 1 之间")
|
|
if rates and "product" not in keys:
|
|
raise ValueError("毛利率映射需要 product 维度")
|
|
if mapping.get("roi_formula") not in ("profit_over_cost", "gross_less_cost_over_cost"):
|
|
raise ValueError("请选择 ROI 口径")
|
|
products = mapping.get("self_operated_products", [])
|
|
if (
|
|
not isinstance(products, list)
|
|
or len(products) > 500
|
|
or any(not isinstance(p, str) or not 1 <= len(p) <= 256 for p in products)
|
|
):
|
|
raise ValueError("自营产品范围无效")
|
|
if products and "product" not in keys:
|
|
raise ValueError("自营产品范围需要 product 维度")
|
|
return mapping
|
|
|
|
|
|
def decimal(value, required=False):
|
|
if value is None or str(value).strip() == "":
|
|
if required:
|
|
raise ValueError("必填金额为空")
|
|
return None
|
|
try:
|
|
number = Decimal(str(value).replace(",", "").strip())
|
|
if not number.is_finite() or abs(number) >= Decimal("1e16"):
|
|
raise ValueError("金额超出范围")
|
|
if number != number.quantize(Decimal("1e-24")):
|
|
raise ValueError("金额精度超过二十四位小数")
|
|
return number
|
|
except InvalidOperation as exc:
|
|
raise ValueError("不是有效金额") from exc
|
|
|
|
|
|
def period_value(values):
|
|
if values.get("period"):
|
|
value = values["period"]
|
|
if isinstance(value, (date, datetime)):
|
|
return value.strftime("%Y-%m")
|
|
text = str(value).strip().replace("/", "-")
|
|
match = re.fullmatch(r"(\d{4})-(\d{1,2})(?:-\d{1,2}(?:T.*)?)?", text)
|
|
if not match:
|
|
raise ValueError("期间必须是 YYYY-MM 或日期")
|
|
year, month = map(int, match.groups())
|
|
else:
|
|
year = int(str(values.get("year", "")).removesuffix("年"))
|
|
month = int(str(values.get("month", "")).removesuffix("月"))
|
|
if not 1900 <= year <= 2200 or not 1 <= month <= 12:
|
|
raise ValueError("期间超出有效范围")
|
|
return f"{year:04d}-{month:02d}"
|
|
|
|
|
|
def process_dataset(dataset_id):
|
|
ds = Dataset.objects.get(pk=dataset_id)
|
|
mapping = ds.mapping
|
|
quality = {
|
|
"negative_revenue": 0,
|
|
"missing_cost": 0,
|
|
"missing_gross": 0,
|
|
"missing_customer": 0,
|
|
"blank_rows": 0,
|
|
"g_l_mismatch": 0,
|
|
"i_n_mismatch": 0,
|
|
}
|
|
names = {c["index"]: c["name"] for c in mapping["columns"]}
|
|
reference_layout = names.get(6) == names.get(11) == "应收金额" and names.get(13) == "实际收入"
|
|
periods, batch, count = set(), [], 0
|
|
try:
|
|
with transaction.atomic():
|
|
if ds.facts.exists():
|
|
raise ValueError("数据版本已包含明细，禁止覆盖")
|
|
for row_no, row in enumerate(rows_for(ds.file.path, mapping["sheet"]), 1):
|
|
if row_no <= mapping["header"]:
|
|
continue
|
|
if not any(v is not None and str(v).strip() for v in row):
|
|
quality["blank_rows"] += 1
|
|
continue
|
|
count += 1
|
|
if count > settings.MAX_IMPORT_ROWS:
|
|
raise ValueError("数据行超过配置上限")
|
|
values, dims = {}, {}
|
|
for col in mapping["columns"]:
|
|
v = row[col["index"]] if col["index"] < len(row) else None
|
|
if col["role"] == "dimension":
|
|
dims[col["key"]] = safe_text(v)[:256] or "（未分类）"
|
|
elif col["role"] != "ignore":
|
|
values[col["role"]] = v
|
|
try:
|
|
period = period_value(values)
|
|
revenue = decimal(values.get("revenue"), required=True)
|
|
nums = {
|
|
k: decimal(values.get(k))
|
|
for k in ("cost", "gross", "quantity", "list_amount", "discount")
|
|
}
|
|
if nums["gross"] is None and dims.get("product") in mapping.get("margin_rates", {}):
|
|
nums["gross"] = (
|
|
revenue * Decimal(str(mapping["margin_rates"][dims["product"]]))
|
|
).quantize(Decimal("1e-24"))
|
|
customer = safe_text(values.get("customer"))[:256] or None
|
|
except (ValueError, TypeError, OverflowError) as exc:
|
|
raise ValueError(f"第 {row_no} 行：{exc}") from exc
|
|
quality["negative_revenue"] += revenue < 0
|
|
if reference_layout and len(row) >= 14:
|
|
quality["g_l_mismatch"] += row[6] != row[11]
|
|
quality["i_n_mismatch"] += row[8] != row[13]
|
|
quality["missing_cost"] += nums["cost"] is None
|
|
quality["missing_gross"] += nums["gross"] is None
|
|
quality["missing_customer"] += customer is None
|
|
periods.add(period)
|
|
batch.append(
|
|
Fact(
|
|
dataset=ds,
|
|
period=period,
|
|
revenue=revenue,
|
|
dimensions=dims,
|
|
customer=customer,
|
|
source_row=row_no,
|
|
raw=list(map(safe_text, row)),
|
|
**nums,
|
|
)
|
|
)
|
|
if len(batch) >= 2000:
|
|
Fact.objects.bulk_create(batch, batch_size=500)
|
|
batch.clear()
|
|
if not count:
|
|
raise ValueError("工作表没有数据行")
|
|
Fact.objects.bulk_create(batch, batch_size=500)
|
|
ds.status, ds.row_count, ds.periods, ds.quality = "ready", count, sorted(periods), quality
|
|
ds.error, ds.finished_at = "", timezone.now()
|
|
ds.save()
|
|
except Exception as exc:
|
|
Dataset.objects.filter(pk=ds.pk).update(
|
|
status="failed",
|
|
error=str(exc)[:500] if isinstance(exc, ValueError) else "导入失败，请查看服务端日志",
|
|
finished_at=timezone.now(),
|
|
)
|
|
if not isinstance(exc, ValueError):
|
|
import logging
|
|
|
|
logging.getLogger(__name__).exception("Import failed for %s", ds.pk)
|
|
return Dataset.objects.get(pk=ds.pk)
|
|
|
|
|
|
def file_digest(file):
|
|
digest = hashlib.sha256()
|
|
for chunk in file.chunks():
|
|
digest.update(chunk)
|
|
file.seek(0)
|
|
return digest.hexdigest()
|