yeuimu
08d26710b4
- src 标准布局：config/providers/pipeline/report + CLI/Tk GUI 双入口 - 级联省钱：豆包全量初筛，仅无违规/违规不明图进 DeepSeek 复检（含两票复核） - 输出：时间戳目录 + 分类文件夹图片归档 + Excel 报表 - 30 个单元测试（tests/，测试图片不入库）
237 lines
9.7 KiB
Python
237 lines
9.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""检测结果输出组织与 Excel 报表生成。
|
||
|
||
输出结构：
|
||
<输出目录>/检测结果_<时间戳>/
|
||
违规检测结果_<时间戳>.xlsx # 一级目录放 Excel
|
||
<违规分类1>/ 图1.jpg 图2.jpg … # 每个分类一个文件夹，图片归档其中
|
||
<违规分类2>/ …
|
||
"""
|
||
import os
|
||
import re
|
||
import shutil
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from openpyxl import Workbook
|
||
from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
|
||
from openpyxl.utils import get_column_letter
|
||
|
||
# 设计令牌（三色纪律：主色 + 语义强调色 + 中性灰）
|
||
PRIMARY = "1B2A4A"
|
||
NEUTRAL_900 = "37352F"
|
||
NEUTRAL_600 = "8C8A84"
|
||
NEUTRAL_200 = "E9E9E8"
|
||
NEUTRAL_100 = "F7F7F5"
|
||
ACCENT_POSITIVE = "1B7D46"
|
||
ACCENT_NEGATIVE = "C0392B"
|
||
ACCENT_WARNING = "D4820A"
|
||
FONT_NAME = "Microsoft YaHei"
|
||
|
||
STATUS_LABEL = {"ok": "", "parse_fail": "模型返回无法解析为标准JSON",
|
||
"error": "调用失败（网络/服务异常）"}
|
||
|
||
|
||
def sanitize_dirname(name: str) -> str:
|
||
"""分类名转合法文件夹名。"""
|
||
s = re.sub(r'[<>:"/\\|?*]', "_", str(name)).strip(" .")
|
||
return s[:80] or "未分类"
|
||
|
||
|
||
def organize_output(rows, src_folder, out_dir, base_name="检测结果") -> Path:
|
||
"""建立时间戳目录与分类文件夹，把图片复制进对应分类，返回时间戳目录。"""
|
||
ts_dir = Path(out_dir) / f"{base_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
|
||
for r in rows:
|
||
cat = sanitize_dirname(r.get("category") or "检测异常")
|
||
d = ts_dir / cat
|
||
d.mkdir(parents=True, exist_ok=True)
|
||
src = Path(src_folder) / r.get("file", "")
|
||
if src.is_file():
|
||
shutil.copy2(src, d / src.name)
|
||
return ts_dir
|
||
|
||
|
||
def _cat_num(cat):
|
||
m = re.match(r"^(\d+)\.", str(cat).strip())
|
||
return int(m.group(1)) if m else (99 if str(cat).strip() != "无违规" else 98)
|
||
|
||
|
||
def _cat_color(cat):
|
||
c = str(cat).strip()
|
||
if c == "无违规":
|
||
return ACCENT_POSITIVE
|
||
if c in ("违规不明", "检测异常"):
|
||
return ACCENT_WARNING
|
||
return ACCENT_NEGATIVE
|
||
|
||
|
||
def _est_lines(text, width):
|
||
"""按列宽估算换行行数（CJK 记 1.7 宽）。"""
|
||
if not text:
|
||
return 1
|
||
per_line = max(int(width / 1.7), 4)
|
||
lines = 0
|
||
for seg in str(text).split("\n"):
|
||
lines += max(1, -(-len(seg) // per_line))
|
||
return lines
|
||
|
||
|
||
def _auto_heights(ws, widths, start_row, end_row):
|
||
for rn in range(start_row, end_row + 1):
|
||
lines = 1
|
||
for col, w in widths.items():
|
||
cell = ws[f"{col}{rn}"]
|
||
if cell.alignment and cell.alignment.wrap_text:
|
||
lines = max(lines, _est_lines(cell.value, w))
|
||
ws.row_dimensions[rn].height = max(22, 20 + (lines - 1) * 15)
|
||
|
||
|
||
def build_report(rows, out_dir, meta=None):
|
||
"""生成 Excel 报表，返回文件路径。rows 来自 pipeline.run_detection。"""
|
||
meta = meta or {}
|
||
wb = Workbook()
|
||
|
||
# ---------- Sheet 1: 检测结果 ----------
|
||
ws = wb.active
|
||
ws.title = "检测结果"
|
||
headers = ["序号", "文件名", "违规分类", "判定通道", "文字/图形属性",
|
||
"侵权/违规逻辑", "票型/复核", "备注"]
|
||
last_col = len(headers) + 1
|
||
widths = {"B": 6, "C": 42, "D": 26, "E": 13, "F": 32, "G": 50, "H": 30, "I": 20}
|
||
|
||
ws.sheet_view.showGridLines = False
|
||
ws.column_dimensions["A"].width = 3
|
||
ws.row_dimensions[1].height = 15
|
||
for col, w in widths.items():
|
||
ws.column_dimensions[col].width = w
|
||
|
||
# 标题
|
||
ws.merge_cells(start_row=2, start_column=2, end_row=2, end_column=last_col)
|
||
t = ws["B2"]
|
||
t.value = meta.get("title", "商品图合规检测工具 - 检测结果")
|
||
t.font = Font(name=FONT_NAME, size=16, bold=True, color=PRIMARY)
|
||
t.alignment = Alignment(horizontal="left", vertical="center")
|
||
ws.row_dimensions[2].height = 32
|
||
ws.row_dimensions[3].height = 8
|
||
|
||
# 表头
|
||
hfill = PatternFill("solid", fgColor=PRIMARY)
|
||
hborder = Border(bottom=Side(style="thin", color=NEUTRAL_200))
|
||
for ci, h in enumerate(headers, start=2):
|
||
c = ws.cell(row=4, column=ci, value=h)
|
||
c.fill = hfill
|
||
c.font = Font(name=FONT_NAME, size=11, bold=True, color="FFFFFF")
|
||
c.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
|
||
c.border = hborder
|
||
ws.row_dimensions[4].height = 28
|
||
|
||
# 数据行（隔行填充）
|
||
for i, r in enumerate(rows):
|
||
rn = 5 + i
|
||
values = [i + 1, r.get("file", ""), r.get("category", ""),
|
||
r.get("channel", ""), r.get("attr", ""), r.get("logic", ""),
|
||
r.get("vote_note", "单票判定（未触发复核）"),
|
||
r.get("remark", STATUS_LABEL.get(r.get("status", ""), ""))]
|
||
fill = PatternFill("solid", fgColor=NEUTRAL_100 if i % 2 else "FFFFFF")
|
||
for ci, v in enumerate(values, start=2):
|
||
c = ws.cell(row=rn, column=ci, value=v)
|
||
c.fill = fill
|
||
c.font = Font(name=FONT_NAME, size=11, color=NEUTRAL_900)
|
||
c.alignment = Alignment(horizontal="left", vertical="center", wrap_text=True)
|
||
ws.cell(row=rn, column=2).alignment = Alignment(horizontal="right", vertical="center")
|
||
ws.cell(row=rn, column=4).font = Font(name=FONT_NAME, size=11,
|
||
color=_cat_color(values[2]))
|
||
vnote = ws.cell(row=rn, column=8)
|
||
if "人工复核" in str(vnote.value):
|
||
vnote.font = Font(name=FONT_NAME, size=11, color=ACCENT_WARNING)
|
||
last_row = 4 + len(rows)
|
||
_auto_heights(ws, widths, 5, last_row)
|
||
ws.freeze_panes = "C5"
|
||
|
||
# 底注
|
||
now = datetime.now().strftime("%Y-%m-%d %H:%M")
|
||
note1 = (f"检测模式：{meta.get('mode', 'cascade')}（豆包初筛 + DeepSeek 复检无违规）｜"
|
||
f"图片总数：{len(rows)}｜生成时间：{now}")
|
||
note2 = ("依据《违规检测提示词》17 类标准判定；DeepSeek temperature=0 + JSON 模式、"
|
||
"max_tokens 可配置；判定为无违规的图片自动追加两票复核（3 票取多数，分歧从严取违规并标记人工复核）。")
|
||
for off, note in ((2, note1), (3, note2)):
|
||
c = ws.cell(row=last_row + off, column=2, value=note)
|
||
c.font = Font(name=FONT_NAME, size=9, color=NEUTRAL_600)
|
||
|
||
# ---------- Sheet 2: 统计汇总 ----------
|
||
ws2 = wb.create_sheet("统计汇总")
|
||
counts = {}
|
||
channels = {}
|
||
for r in rows:
|
||
c = str(r.get("category", "")).strip() or "（空）"
|
||
counts[c] = counts.get(c, 0) + 1
|
||
ch = r.get("channel", "")
|
||
channels[ch] = channels.get(ch, 0) + 1
|
||
ordered = sorted(counts.items(), key=lambda kv: _cat_num(kv[0]))
|
||
|
||
widths2 = {"B": 46, "C": 10, "D": 10}
|
||
ws2.sheet_view.showGridLines = False
|
||
ws2.column_dimensions["A"].width = 3
|
||
ws2.row_dimensions[1].height = 15
|
||
for col, w in widths2.items():
|
||
ws2.column_dimensions[col].width = w
|
||
ws2.merge_cells("B2:D2")
|
||
t2 = ws2["B2"]
|
||
t2.value = "违规分类统计汇总"
|
||
t2.font = Font(name=FONT_NAME, size=16, bold=True, color=PRIMARY)
|
||
t2.alignment = Alignment(horizontal="left", vertical="center")
|
||
ws2.row_dimensions[2].height = 32
|
||
ws2.row_dimensions[3].height = 8
|
||
|
||
for ci, h in enumerate(["违规分类", "数量", "占比"], start=2):
|
||
c = ws2.cell(row=4, column=ci, value=h)
|
||
c.fill = hfill
|
||
c.font = Font(name=FONT_NAME, size=11, bold=True, color="FFFFFF")
|
||
c.alignment = Alignment(horizontal="center", vertical="center")
|
||
c.border = hborder
|
||
ws2.row_dimensions[4].height = 28
|
||
|
||
total = len(rows)
|
||
for i, (cat, n) in enumerate(ordered):
|
||
rn = 5 + i
|
||
fill = PatternFill("solid", fgColor=NEUTRAL_100 if i % 2 else "FFFFFF")
|
||
vals = [cat, n, (n / total if total else 0)]
|
||
for ci, v in enumerate(vals, start=2):
|
||
c = ws2.cell(row=rn, column=ci, value=v)
|
||
c.fill = fill
|
||
c.font = Font(name=FONT_NAME, size=11, color=NEUTRAL_900)
|
||
c.alignment = Alignment(horizontal="left" if ci == 2 else "right",
|
||
vertical="center")
|
||
ws2.cell(row=rn, column=3).number_format = "#,##0"
|
||
ws2.cell(row=rn, column=4).number_format = "0.0%"
|
||
ws2.cell(row=rn, column=2).font = Font(name=FONT_NAME, size=11,
|
||
color=_cat_color(cat))
|
||
ws2.row_dimensions[rn].height = 22
|
||
|
||
trow = 5 + len(ordered)
|
||
ws2.cell(row=trow, column=2, value="合计")
|
||
ws2.cell(row=trow, column=3, value=total).number_format = "#,##0"
|
||
ws2.cell(row=trow, column=4, value=1.0 if total else 0).number_format = "0.0%"
|
||
for ci in range(2, 5):
|
||
c = ws2.cell(row=trow, column=ci)
|
||
c.fill = PatternFill("solid", fgColor="D6E4F0")
|
||
c.font = Font(name=FONT_NAME, size=11, bold=True, color=PRIMARY)
|
||
c.border = Border(top=Side(style="medium", color=NEUTRAL_200))
|
||
c.alignment = Alignment(horizontal="left" if ci == 2 else "right", vertical="center")
|
||
ws2.row_dimensions[trow].height = 26
|
||
|
||
ch_txt = "、".join(f"{k} {v} 张" for k, v in channels.items())
|
||
cost_txt = ""
|
||
if meta.get("ds_peak_cost") is not None:
|
||
cost_txt = (f"｜DeepSeek {meta.get('ds_calls', 0)} 次调用成本≈"
|
||
f"¥{meta['ds_peak_cost']:.2f}（高峰）/¥{meta['ds_idle_cost']:.2f}（空闲）")
|
||
c = ws2.cell(row=trow + 2, column=2,
|
||
value=f"判定通道分布：{ch_txt}{cost_txt}｜生成时间：{now}")
|
||
c.font = Font(name=FONT_NAME, size=9, color=NEUTRAL_600)
|
||
|
||
wb.properties.creator = "violation-detector"
|
||
os.makedirs(out_dir, exist_ok=True)
|
||
out = os.path.join(out_dir, f"违规检测结果_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx")
|
||
wb.save(out)
|
||
return out
|