Skip to content

S
skills
Commit 62507a1d authored by allen.wang's avatar allen.wang
Browse files
Options

feat:init

parent 594861cb
Show whitespace changes
Inline Side-by-side
Showing with 1676 additions and 0 deletions
vip-report/bin/vip-report-campaign-s11-sync.ps1 0 → 100644
View file @ 62507a1d
<#
用途：
- 同步 S11（Campaign-年货节）来源数据，落盘校验结果并生成 render-ops。
- 可选直接渲染 PPT 并执行 compare。
#>
param(
[string]$ConfigPath = "C:\Users\niuniu\.codex\vip-report\config.yaml",
[string]$Slides = "S11",
[string]$ReportMonth = "",
[int]$ReportYear = 0,
[int]$CompareYear = 0,
[int]$ShopId = 20,
[switch]$DisableStrictJanuaryCheck,
[switch]$Render,
[string]$OutputPath = "C:\workspace\cursor\output\vip-report\generated-campaign-s11-live.pptx",
[string]$CompareOutputPath = "C:\workspace\cursor\output\vip-report\generated-campaign-s11-live.compare.json"
)
$ErrorActionPreference = "Stop"
$root = Split-Path -Parent $PSScriptRoot
$opsPath = "C:\workspace\cursor\output\vip-report\render-ops.campaign-s11.live.json"
$templatePath = "C:\Users\niuniu\Desktop\Report.pptx"
$pythonArgs = @(
"$root\scripts\sync_campaign_s11_assets.py",
"--config", "$ConfigPath",
"--slides", "$Slides",
"--shop-id", "$ShopId"
)
if ($ReportMonth) {
$pythonArgs += @("--report-month", "$ReportMonth")
}
if ($ReportYear -gt 0) {
$pythonArgs += @("--report-year", "$ReportYear")
}
if ($CompareYear -gt 0) {
$pythonArgs += @("--compare-year", "$CompareYear")
}
if ($DisableStrictJanuaryCheck) {
$pythonArgs += @("--disable-strict-january-check")
}
python @pythonArgs
if ($LASTEXITCODE -ne 0) {
throw "sync_campaign_s11_assets.py failed with exit code $LASTEXITCODE"
}
if ($Render) {
powershell -ExecutionPolicy Bypass -File "$root\bin\vip-report-render.ps1" -TemplatePath "$templatePath" -OutputPath "$OutputPath" -OperationsPath "$opsPath" | Out-Null
if ($LASTEXITCODE -ne 0) {
throw "vip-report-render.ps1 failed with exit code $LASTEXITCODE"
}
python "$root\scripts\compare_pptx.py" "$templatePath" "$OutputPath" --output "$CompareOutputPath"
if ($LASTEXITCODE -ne 0) {
throw "compare_pptx.py failed with exit code $LASTEXITCODE"
}
Write-Output $OutputPath
Write-Output $CompareOutputPath
} else {
Write-Output $opsPath
}
vip-report/scripts/sync_campaign_s11_assets.py 0 → 100644
View file @ 62507a1d
"""VIP 报表数据同步：采集 S11（Campaign-年货节）来源并落盘校验结果。"""
from __future__ import annotations
import argparse
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
import io
import json
import re
from pathlib import Path
from typing import Any
from urllib.parse import quote
from urllib.request import Request, urlopen
import pymysql
import yaml
from PIL import Image, ImageDraw, ImageFont
import matplotlib.pyplot as plt
from matplotlib.lines import Line2D
import numpy as np
# 中文月份归一化映射：CLI/config 可传中文或数字，最终统一为 month_num。
MONTH_CN_MAP = {
1: "一月",
2: "二月",
3: "三月",
4: "四月",
5: "五月",
6: "六月",
7: "七月",
8: "八月",
9: "九月",
10: "十月",
11: "十一月",
12: "十二月",
}
MONTH_CN_TO_NUM = {label: number for number, label in MONTH_CN_MAP.items()}
LEVEL1_ORDER = ["BAGS", "SHOES", "ACC"]
ACTIVITY_MEMOS = {"活动款", "活动折扣款"}
NON_ACTIVITY_ONLY_MEMOS = {"非活动款"}
ESS_MEMO = "ESS"
TABLE3_CATEGORY_ORDER = [
"女单鞋",
"女凉鞋",
"女拖鞋",
"女休闲鞋",
"女靴",
"单肩包",
"卡包",
"钱包",
"手拿包",
"手提包",
"双肩包",
"胸包/腰包",
]
TABLE3_CATEGORY_SET = set(TABLE3_CATEGORY_ORDER)
TABLE3_BAGS_CATEGORY_SET = {"单肩包", "卡包", "钱包", "手拿包", "手提包", "双肩包", "胸包/腰包"}
TABLE3_SHOES_CATEGORY_SET = {"女单鞋", "女凉鞋", "女拖鞋", "女休闲鞋", "女靴"}
# 明细 fallback：当同比期没有可直接映射的 article 时，按 level3 兜底归类。
DETAIL_LEVEL3_FALLBACK = {
"TOP HANDLE": "手提包",
"BACKPACK": "双肩包",
"BUM / BELT BAG": "胸包/腰包",
"CARD HOLDER": "卡包",
"WALLET": "钱包",
"COMPACT WALLET": "钱包",
"PHONE POUCH": "钱包",
"MINI PURSE": "钱包",
"POUCH": "手拿包",
"CLUTCH": "手拿包",
"WALLET ON CHAIN": "手拿包",
"SHOULDER BAG": "单肩包",
"SHOULDER": "单肩包",
"CROSSBODY": "单肩包",
"HOBO": "单肩包",
"TOTE": "单肩包",
"BUCKET": "单肩包",
"CHAIN BAG": "单肩包",
"MINI BAG": "单肩包",
"MICRO BAG": "单肩包",
"HEELS": "女单鞋",
"FLATS": "女单鞋",
"WEDGES": "女凉鞋",
"SNEAKERS": "女休闲鞋",
"BOOTS": "女靴",
"ACCESSORY": "手拿包",
}
DETAIL_CATEGORY_DEFAULT = "手拿包"
JAN_TEMPLATE_EXPECTED = {
"sales_10k": 910,
"activity_ratio_qty": "11:89",
}
S11_NARRATIVE_SHAPE = {
"slide": 11,
"shape_id": 27,
"shape_name": "TextBox 15",
}
S11_TABLE1_SHAPE = {
"slide": 11,
"shape_id": 2,
"shape_name": "表格 1",
}
S11_TABLE2_SHAPE = {
"slide": 11,
"shape_id": 3,
"shape_name": "表格 2",
}
S11_TABLE3_SHAPE = {
"slide": 11,
"shape_id": 4,
"shape_name": "表格 3",
}
S11_CHART_QTY_SHAPE = {
"slide": 11,
"shape_id": 28,
"shape_name": "图表 27",
}
S11_CHART_SALES_SHAPE = {
"slide": 11,
"shape_id": 29,
"shape_name": "图表 28",
}
S11_TOP10_SHAPE = {
"slide": 11,
"shape_id": 5,
"shape_name": "图片 4",
}
@dataclass(frozen=True)
class ReportPeriod:
"""报表期对象，统一管理当期与同比期字符串口径。"""
month_label: str
month_num: int
report_year: int
compare_year: int
@property
def ym(self) -> str:
return f"{self.report_year}-{self.month_num:02d}"
@property
def compare_ym(self) -> str:
return f"{self.compare_year}-{self.month_num:02d}"
@dataclass(frozen=True)
class CampaignWindow:
"""活动窗口对象：用于统一 S11 的时间口径与标题展示。"""
current_start: date
current_end: date
compare_start: date
compare_end: date
@property
def label(self) -> str:
return f"{self.current_start.month}.{self.current_start.day}-{self.current_end.month}.{self.current_end.day}"
def normalize_month_label(raw_month: Any) -> str:
"""把 1/01/1月/一月 等输入统一为中文月份，避免 SQL 组装分支膨胀。"""
if raw_month is None:
return "一月"
if isinstance(raw_month, int):
return MONTH_CN_MAP.get(raw_month, "一月")
text = str(raw_month).strip()
if text in MONTH_CN_TO_NUM:
return text
match = re.fullmatch(r"0?([1-9]|1[0-2])(?:月)?", text)
if match:
return MONTH_CN_MAP[int(match.group(1))]
return text
def month_label_to_number(month_label: str) -> int:
return MONTH_CN_TO_NUM.get(month_label, 1)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="同步 S11 Campaign 数据来源并生成校验产物。")
parser.add_argument(
"--config",
default=r"C:\Users\niuniu\.codex\vip-report\config.yaml",
help="VIP report config.yaml 路径",
)
parser.add_argument(
"--slides",
default="S11",
help="仅用于保持统一参数风格，当前仅支持 S11",
)
parser.add_argument(
"--report-month",
default="",
help="报表月份（中文/数字）",
)
parser.add_argument(
"--report-year",
type=int,
default=0,
help="报表年份",
)
parser.add_argument(
"--compare-year",
type=int,
default=0,
help="同比年份",
)
parser.add_argument(
"--shop-id",
type=int,
default=20,
help="店铺 id，默认 20（CKC-VIP）",
)
parser.add_argument(
"--disable-strict-january-check",
action="store_true",
help="关闭 2026-01 模板口径强校验。",
)
return parser.parse_args()
def resolve_report_period(args: argparse.Namespace, config: dict[str, Any]) -> ReportPeriod:
"""周期优先级：CLI > config.yaml > 默认值。"""
report_cfg = config.get("report", {})
month_label = normalize_month_label(args.report_month or report_cfg.get("month_cn", "一月"))
month_num = month_label_to_number(month_label)
report_year = int(args.report_year or report_cfg.get("year", 2026))
compare_year = int(args.compare_year or report_cfg.get("compare_year", report_year - 1))
return ReportPeriod(
month_label=month_label,
month_num=month_num,
report_year=report_year,
compare_year=compare_year,
)
def parse_ymd(raw: Any) -> date | None:
if raw is None:
return None
text = str(raw).strip()
if not text:
return None
try:
return datetime.strptime(text[:10], "%Y-%m-%d").date()
except ValueError:
return None
def to_decimal(value: Any) -> Decimal:
if value is None:
return Decimal("0")
if isinstance(value, Decimal):
return value
text = str(value).strip()
if not text:
return Decimal("0")
try:
return Decimal(text)
except InvalidOperation:
return Decimal("0")
def to_int(value: Any) -> int:
if value is None:
return 0
try:
return int(value)
except (TypeError, ValueError):
return 0
def round_half_up(value: Decimal) -> int:
return int(value.quantize(Decimal("1"), rounding=ROUND_HALF_UP))
def safe_ratio(cur: Decimal, prev: Decimal) -> Decimal:
if prev == 0:
return Decimal("0")
return (cur - prev) / prev
def ratio_string(left: int, right: int) -> str:
"""把两段量转成展示口径（如 11:89），供文案和校验共用。"""
total = left + right
if total <= 0:
return "0:0"
left_pct = round_half_up(Decimal(left) * Decimal("100") / Decimal(total))
right_pct = 100 - left_pct
return f"{left_pct}:{right_pct}"
def ratio_percent_text(cur: Decimal, prev: Decimal) -> str:
pct = round_half_up(safe_ratio(cur, prev) * Decimal("100"))
return f"{pct}%"
def pct_of_total_text(value: Decimal, total: Decimal) -> str:
if total == 0:
return "0%"
pct = round_half_up(value * Decimal("100") / total)
return f"{pct}%"
def format_int_with_comma(value: int) -> str:
return f"{value:,}"
def format_plain_int(value: int) -> str:
return str(value)
def format_sales_2_with_comma(value: Decimal) -> str:
return f"{value.quantize(Decimal('0.01')):,.2f}"
def format_sales_int_with_comma(value: Decimal) -> str:
return f"{round_half_up(value):,}"
def format_sales_2_plain(value: Decimal) -> str:
return f"{value.quantize(Decimal('0.01')):.2f}"
def format_sales_trim_plain(value: Decimal) -> str:
normalized = value.quantize(Decimal("0.01")).normalize()
text = format(normalized, "f")
if "." in text:
text = text.rstrip("0").rstrip(".")
return text
def fetch_one(cursor: pymysql.cursors.Cursor, sql: str, params: tuple[Any, ...]) -> dict[str, Any]:
cursor.execute(sql, params)
row = cursor.fetchone() or {}
return dict(row)
def fetch_all(cursor: pymysql.cursors.Cursor, sql: str, params: tuple[Any, ...]) -> list[dict[str, Any]]:
cursor.execute(sql, params)
rows = cursor.fetchall() or []
return [dict(row) for row in rows]
def infer_campaign_window(shop_rows: list[dict[str, Any]], period: ReportPeriod) -> CampaignWindow:
"""从 oms_shop_report 行里推导活动窗口：
- activitybegin 视为当期起始；
- activityend 视为同比期结束；
- 当期结束使用 activityend 的月日 + report_year 重建。
"""
start_candidates = [parse_ymd(item.get("activitybegin")) for item in shop_rows]
end_candidates = [parse_ymd(item.get("activityend")) for item in shop_rows]
starts = [item for item in start_candidates if item is not None]
ends = [item for item in end_candidates if item is not None]
if not starts or not ends:
raise SystemExit("oms_shop_report 缺少 activitybegin/activityend，无法推导 S11 活动窗口。")
current_start_seed = min(starts)
end_seed = max(ends)
current_start = date(period.report_year, current_start_seed.month, current_start_seed.day)
compare_start = date(period.compare_year, current_start_seed.month, current_start_seed.day)
current_end = date(period.report_year, end_seed.month, end_seed.day)
compare_end = date(period.compare_year, end_seed.month, end_seed.day)
# 若 activityend 本身就是同比年，沿用它作为 compare_end 更稳妥。
if end_seed.year == period.compare_year:
compare_end = end_seed
current_end = date(period.report_year, end_seed.month, end_seed.day)
elif end_seed.year == period.report_year:
current_end = end_seed
compare_end = date(period.compare_year, end_seed.month, end_seed.day)
return CampaignWindow(
current_start=current_start,
current_end=current_end,
compare_start=compare_start,
compare_end=compare_end,
)
def load_s11_narrative_template(workdir: Path) -> str:
"""从 shape-inventory 读取 S11 文案模板；读取失败时回退到内置文案骨架。"""
inventory_path = workdir / "shape-inventory.json"
if inventory_path.exists():
try:
inventory = json.loads(inventory_path.read_text(encoding="utf-8-sig"))
for slide in inventory.get("Slides", []):
if slide.get("Slide") != S11_NARRATIVE_SHAPE["slide"]:
continue
for shape in slide.get("Shapes", []):
if shape.get("Id") == S11_NARRATIVE_SHAPE["shape_id"]:
text = shape.get("Text") or ""
if isinstance(text, str) and text.strip():
return text
except (json.JSONDecodeError, OSError):
pass
return (
"唯品年货节销售910w，于1.24起销售逐渐回正并反超，同比增幅在11%。"
"除客单仍较去年低3%外，流量及转化均呈现正向增长。流量在联投帮助下涨幅更加明显；\r"
"活动款及非活动款比例为11:89。从品类层面来看，鞋包销量同比均为正向。"
"鞋类增幅远大于包，达41%。主要依靠独家款及正价RP凉鞋及单鞋驱动。"
)
def rewrite_s11_narrative(
template_text: str,
*,
sales_10k: int,
sales_lfl_pct: int,
atv_lfl_pct: int,
activity_ratio_qty: str,
shoes_sales_lfl_pct: int,
) -> str:
"""按固定槽位替换 S11 文案核心数字，确保版式与语气保持模板风格。"""
text = template_text
text = re.sub(r"销售\d+w", f"销售{sales_10k}w", text, count=1)
sales_phrase = f"同比增幅在{sales_lfl_pct}%" if sales_lfl_pct >= 0 else f"同比降幅在{abs(sales_lfl_pct)}%"
text = re.sub(r"同比[增降]幅在-?\d+%", sales_phrase, text, count=1)
if atv_lfl_pct > 0:
atv_phrase = f"客单仍较去年高{atv_lfl_pct}%"
text = re.sub(r"客单仍较去年[低高]-?\d+%", atv_phrase, text, count=1)
elif atv_lfl_pct < 0:
atv_phrase = f"客单仍较去年低{abs(atv_lfl_pct)}%"
text = re.sub(r"客单仍较去年[低高]-?\d+%", atv_phrase, text, count=1)
else:
text = re.sub(r"客单仍较去年[低高]-?\d+%", "客单与去年持平", text, count=1)
text = re.sub(r"比例为\d+:\d+", f"比例为{activity_ratio_qty}", text, count=1)
shoes_phrase = f"鞋类增幅远大于包，达{shoes_sales_lfl_pct}%"
if shoes_sales_lfl_pct < 0:
shoes_phrase = f"鞋类降幅明显，达{abs(shoes_sales_lfl_pct)}%"
text = re.sub(r"鞋类增幅远大于包，达-?\d+%", shoes_phrase, text, count=1)
return text
def build_table_cells(rows: list[list[str]]) -> list[dict[str, Any]]:
cells: list[dict[str, Any]] = []
for row_index, row in enumerate(rows, start=1):
for col_index, value in enumerate(row, start=1):
cells.append(
{
"row": row_index,
"col": col_index,
"new_text": value,
}
)
return cells
def choose_cn_category_for_detail(
*,
articlecode: str,
level3: str,
level4: str,
article_to_cn: dict[str, str],
combo_to_cn: dict[tuple[str, str], str],
level3_to_cn: dict[str, str],
) -> str:
# 优先级：同款 article 映射 > level3+level4 > level3 > fallback。
if articlecode in article_to_cn and article_to_cn[articlecode] in TABLE3_CATEGORY_SET:
return article_to_cn[articlecode]
if (level3, level4) in combo_to_cn and combo_to_cn[(level3, level4)] in TABLE3_CATEGORY_SET:
return combo_to_cn[(level3, level4)]
if level3 in level3_to_cn and level3_to_cn[level3] in TABLE3_CATEGORY_SET:
return level3_to_cn[level3]
fallback = DETAIL_LEVEL3_FALLBACK.get(level3, DETAIL_CATEGORY_DEFAULT)
if fallback not in TABLE3_CATEGORY_SET:
return DETAIL_CATEGORY_DEFAULT
return fallback
def load_font(size: int, *, bold: bool = False) -> ImageFont.FreeTypeFont | ImageFont.ImageFont:
"""优先加载系统中文字体，保证导出的 TOP10 图片中文不乱码。"""
candidates: list[str] = []
if bold:
candidates.extend(
[
r"C:\Windows\Fonts\msyhbd.ttc",
r"C:\Windows\Fonts\simhei.ttf",
]
)
candidates.extend(
[
r"C:\Windows\Fonts\msyh.ttc",
r"C:\Windows\Fonts\simsun.ttc",
r"C:\Windows\Fonts\arial.ttf",
]
)
for path in candidates:
try:
return ImageFont.truetype(path, size=size)
except OSError:
continue
return ImageFont.load_default()
def measure_text(draw: ImageDraw.ImageDraw, text: str, font: ImageFont.ImageFont) -> tuple[int, int]:
bbox = draw.textbbox((0, 0), text, font=font)
return bbox[2] - bbox[0], bbox[3] - bbox[1]
def candidate_image_codes(article: str) -> list[str]:
"""按款号生成候选图片编码，优先全码，再尝试去掉短尾缀。"""
article = article.strip()
if not article:
return []
codes = [article]
suffix_match = re.fullmatch(r"(.+)-([A-Za-z0-9]{1,2})", article)
if suffix_match:
base_code = suffix_match.group(1)
if base_code not in codes:
codes.append(base_code)
return codes
def load_article_image(article: str, *, image_cache_dir: Path) -> Image.Image | None:
"""从 api.charleskeith.cn 下载并缓存商品图，失败则返回 None。"""
image_cache_dir.mkdir(parents=True, exist_ok=True)
for code in candidate_image_codes(article):
cache_path = image_cache_dir / f"{code}.jpg"
if cache_path.exists():
try:
return Image.open(cache_path).convert("RGB")
except OSError:
cache_path.unlink(missing_ok=True)
url = f"https://api.charleskeith.cn/img/{quote(code)}.jpg"
request = Request(url, headers={"User-Agent": "Mozilla/5.0"})
try:
with urlopen(request, timeout=8) as response:
content_type = (response.headers.get("Content-Type") or "").lower()
body = response.read()
if not body or "image" not in content_type:
continue
cache_path.write_bytes(body)
return Image.open(io.BytesIO(body)).convert("RGB")
except Exception:
continue
return None
def render_top10_panel_image(top_rows: list[dict[str, Any]], output_path: Path, *, image_cache_dir: Path) -> None:
"""生成 S11 右侧 TOP10 图块（按当期数据库排序结果）。"""
width = 1000
height = 1980
margin = 14
header_h = 46
row_h = (height - margin * 2 - header_h) // 10
# TOP10 / PIC / QTY / SALES / 活动款备注 / 独家
col_widths = [250, 205, 100, 150, 205, 90]
bg = Image.new("RGB", (width, height), "#F7F7F7")
draw = ImageDraw.Draw(bg)
line_color = "#6F74D8"
header_bg = "#7F83E6"
header_text = "#FFFFFF"
text_color = "#2E2E2E"
light_box = "#EAEAEA"
font_header = load_font(24, bold=True)
font_body = load_font(22, bold=False)
font_small = load_font(18, bold=False)
font_tiny = load_font(16, bold=False)
# 外框与标题栏
draw.rectangle((margin, margin, width - margin, height - margin), outline=line_color, width=3)
draw.rectangle((margin, margin, width - margin, margin + header_h), fill=header_bg, outline=line_color, width=2)
headers = ["TOP10", "PIC", "QTY", "SALES", "活动款备注", "独家"]
x = margin
for idx, title in enumerate(headers):
w = col_widths[idx]
tw, th = measure_text(draw, title, font_header)
draw.text((x + (w - tw) / 2, margin + (header_h - th) / 2 - 2), title, font=font_header, fill=header_text)
x += w
if idx < len(headers) - 1:
draw.line((x, margin, x, height - margin), fill=line_color, width=2)
# 数据行
y0 = margin + header_h
for row_idx in range(10):
top = y0 + row_idx * row_h
bottom = top + row_h
draw.line((margin, bottom, width - margin, bottom), fill=line_color, width=1)
record = top_rows[row_idx] if row_idx < len(top_rows) else None
if not record:
continue
article = str(record.get("article") or "")
qty = to_int(record.get("qty"))
sales = to_decimal(record.get("sales"))
memo = str(record.get("memo") or "")
single = "Y" if str(record.get("single") or "").upper() == "Y" else ""
# TOP10 列：article 编码 + 排名
col0_left = margin
col0_w = col_widths[0]
rank_text = f"#{row_idx + 1}"
draw.text((col0_left + 8, top + 6), rank_text, font=font_small, fill="#666666")
draw.text((col0_left + 8, top + 34), article, font=font_body, fill=text_color)
# PIC 列：按 article 从固定图片域名拉图，失败时再回落占位。
col1_left = margin + col_widths[0]
col1_w = col_widths[1]
box = (
col1_left + 10,
top + 12,
col1_left + col1_w - 10,
bottom - 12,
)
draw.rectangle(box, fill=light_box, outline="#C8C8C8", width=1)
image = load_article_image(article, image_cache_dir=image_cache_dir)
if image is not None:
box_w = box[2] - box[0]
box_h = box[3] - box[1]
scale = min(box_w / max(image.width, 1), box_h / max(image.height, 1))
resized_w = max(1, int(image.width * scale))
resized_h = max(1, int(image.height * scale))
resized = image.resize((resized_w, resized_h), Image.Resampling.LANCZOS)
paste_left = box[0] + (box_w - resized_w) // 2
paste_top = box[1] + (box_h - resized_h) // 2
bg.paste(resized, (paste_left, paste_top))
draw.rectangle(box, outline="#C8C8C8", width=1)
else:
placeholder = "NO PIC"
tw, th = measure_text(draw, placeholder, font_tiny)
draw.text(
(col1_left + (col1_w - tw) / 2, top + (row_h - th) / 2),
placeholder,
font=font_tiny,
fill="#8A8A8A",
)
# 其余列：QTY / SALES / 活动款备注 / 独家
col2_left = margin + col_widths[0] + col_widths[1]
col3_left = col2_left + col_widths[2]
col4_left = col3_left + col_widths[3]
col5_left = col4_left + col_widths[4]
draw.text((col2_left + 8, top + (row_h - 24) / 2), format_plain_int(qty), font=font_body, fill=text_color)
draw.text((col3_left + 8, top + (row_h - 24) / 2), format_sales_trim_plain(sales), font=font_body, fill=text_color)
draw.text((col4_left + 8, top + (row_h - 24) / 2), memo, font=font_body, fill=text_color)
draw.text((col5_left + 26, top + (row_h - 24) / 2), single, font=font_body, fill=text_color)
output_path.parent.mkdir(parents=True, exist_ok=True)
bg.save(output_path)
def nice_percent_axis_bounds(values_pct: list[float], *, floor_min: float = -10.0, ceil_max: float = 50.0) -> tuple[float, float]:
"""给 LFL 百分比轴做温和的上下界，保证点位清晰且不被截断。"""
if not values_pct:
return floor_min, ceil_max
min_v = min(values_pct)
max_v = max(values_pct)
lower = min(floor_min, np.floor((min_v - 2.0) / 5.0) * 5.0)
upper = max(ceil_max, np.ceil((max_v + 2.0) / 5.0) * 5.0)
if upper - lower < 20:
upper = lower + 20
return float(lower), float(upper)
def render_lfl_chart_image(
*,
output_path: Path,
title: str,
current_values: list[float],
compare_values: list[float],
lfl_values: list[float],
) -> None:
"""生成 S11 的 LFL 图（柱形 + 红点，不连线）。"""
categories = ["BAGS", "SHOES"]
x = np.arange(len(categories))
width = 0.32
fig, ax1 = plt.subplots(figsize=(6.5, 3.5), dpi=220)
fig.patch.set_facecolor("white")
ax1.set_facecolor("white")
color_cur = "#6E76E8"
color_prev = "#F1B7DF"
color_lfl = "#FF2D2D"
bars_cur = ax1.bar(x - width / 2, current_values, width, color=color_cur, label="Y-26")
bars_prev = ax1.bar(x + width / 2, compare_values, width, color=color_prev, label="Y-25")
ax1.set_xticks(x)
ax1.set_xticklabels(categories, fontsize=9)
ax1.tick_params(axis="y", labelsize=8, colors="#6B6B6B")
ax1.tick_params(axis="x", labelsize=8, colors="#6B6B6B")
ax1.spines["top"].set_visible(False)
ax1.spines["right"].set_visible(False)
ax1.spines["left"].set_color("#D0D0D0")
ax1.spines["bottom"].set_color("#D0D0D0")
ax1.yaxis.grid(False)
ax1.set_title(title, fontsize=12, fontweight="bold", color="#3F3F3F", pad=10)
ax2 = ax1.twinx()
pct_values = [item * 100.0 for item in lfl_values]
lower, upper = nice_percent_axis_bounds(pct_values)
ax2.set_ylim(lower, upper)
ax2.tick_params(axis="y", labelsize=8, colors="#6B6B6B")
ax2.spines["top"].set_visible(False)
ax2.spines["left"].set_visible(False)
ax2.spines["right"].set_color("#D0D0D0")
# 关键：仅画红点，不画连线。
ax2.scatter(x, pct_values, color=color_lfl, s=28, zorder=6, label="LFL")
for idx, pct in enumerate(pct_values):
ax2.text(
x[idx] + 0.05,
pct,
f"{round(pct):.0f}%",
color="#4A4A4A",
fontsize=9,
va="center",
ha="left",
)
# 右侧百分比轴标签
ticks = np.linspace(lower, upper, num=7)
ax2.set_yticks(ticks)
ax2.set_yticklabels([f"{int(round(v))}%" for v in ticks])
legend_handles = [
bars_cur[0],
bars_prev[0],
Line2D([0], [0], marker="o", color="w", markerfacecolor=color_lfl, markersize=6, label="LFL"),
]
ax1.legend(
handles=legend_handles,
labels=["Y-26", "Y-25", "LFL"],
loc="lower center",
bbox_to_anchor=(0.5, -0.28),
ncol=3,
frameon=False,
fontsize=8,
handlelength=1.2,
columnspacing=0.8,
)
output_path.parent.mkdir(parents=True, exist_ok=True)
fig.tight_layout()
fig.savefig(output_path, facecolor="white")
plt.close(fig)
def run_sync(args: argparse.Namespace) -> dict[str, Any]:
"""S11 同步主流程：
1) 拉取 oms_shop_report / oms_daily_report / oms_daily_report_detail；
2) 计算表格、图表、文案所需全部指标；
3) 对一月模板做强校验；
4) 输出 manifest 与 render-ops。
"""
config_path = Path(args.config)
config = yaml.safe_load(config_path.read_text(encoding="utf-8"))
period = resolve_report_period(args, config)
if "S11" not in {item.strip().upper() for item in args.slides.split(",") if item.strip()}:
raise SystemExit("当前脚本只支持 S11，请传 --slides S11")
mysql_cfg = config["mysql"]
workdir = Path(config["paths"]["workdir"]).resolve()
data_dir = workdir / "data" / "campaign-s11"
data_dir.mkdir(parents=True, exist_ok=True)
conn = pymysql.connect(
host=mysql_cfg["host"],
port=int(mysql_cfg["port"]),
user=mysql_cfg["username"],
password=mysql_cfg["password"],
database=mysql_cfg["database"],
charset="utf8mb4",
cursorclass=pymysql.cursors.DictCursor,
)
try:
with conn.cursor() as cursor:
shop_rows = fetch_all(
cursor,
"""
SELECT article, qty, sales, level1, level3, memo, single, date, activitybegin, activityend
FROM oms_shop_report
WHERE date = %s
""",
(period.ym,),
)
compare_shop_rows = fetch_all(
cursor,
"""
SELECT article, qty, sales, level1, level3, memo, single, date
FROM oms_shop_report
WHERE date = %s
""",
(period.compare_ym,),
)
if not shop_rows:
raise SystemExit(f"oms_shop_report 未找到 {period.ym} 数据。")
campaign_window = infer_campaign_window(shop_rows, period)
summary_cur = fetch_one(
cursor,
"""
SELECT
SUM(COALESCE(dailysales, 0)) AS sales,
SUM(COALESCE(uv, 0)) AS uv,
SUM(COALESCE(tran, 0)) AS tran,
SUM(COALESCE(qtysold, 0)) AS qty
FROM oms_daily_report
WHERE shopid = %s
AND daily BETWEEN %s AND %s
""",
(
args.shop_id,
campaign_window.current_start.strftime("%Y-%m-%d"),
campaign_window.current_end.strftime("%Y-%m-%d"),
),
)
summary_prev = fetch_one(
cursor,
"""
SELECT
SUM(COALESCE(dailysales, 0)) AS sales,
SUM(COALESCE(uv, 0)) AS uv,
SUM(COALESCE(tran, 0)) AS tran,
SUM(COALESCE(qtysold, 0)) AS qty
FROM oms_daily_report
WHERE shopid = %s
AND daily BETWEEN %s AND %s
""",
(
args.shop_id,
campaign_window.compare_start.strftime("%Y-%m-%d"),
campaign_window.compare_end.strftime("%Y-%m-%d"),
),
)
detail_cur_positive = fetch_all(
cursor,
"""
SELECT
articlecode,
level1,
level3,
level4,
SUM(CASE WHEN qty > 0 THEN qty ELSE 0 END) AS qty,
SUM(CASE WHEN sales > 0 THEN sales ELSE 0 END) AS sales
FROM oms_daily_report_detail
WHERE shopid = %s
AND daily BETWEEN %s AND %s
GROUP BY articlecode, level1, level3, level4
HAVING qty > 0 OR sales > 0
""",
(
args.shop_id,
campaign_window.current_start.strftime("%Y-%m-%d"),
campaign_window.current_end.strftime("%Y-%m-%d"),
),
)
detail_prev_positive = fetch_all(
cursor,
"""
SELECT
articlecode,
level1,
level3,
level4,
SUM(CASE WHEN qty > 0 THEN qty ELSE 0 END) AS qty,
SUM(CASE WHEN sales > 0 THEN sales ELSE 0 END) AS sales
FROM oms_daily_report_detail
WHERE shopid = %s
AND daily BETWEEN %s AND %s
GROUP BY articlecode, level1, level3, level4
HAVING qty > 0 OR sales > 0
""",
(
args.shop_id,
campaign_window.compare_start.strftime("%Y-%m-%d"),
campaign_window.compare_end.strftime("%Y-%m-%d"),
),
)
finally:
conn.close()
total_qty = 0
total_sales = Decimal("0")
activity_qty = 0
activity_sales = Decimal("0")
non_activity_only_qty = 0
non_activity_only_sales = Decimal("0")
ess_qty = 0
ess_sales = Decimal("0")
exclusive_qty = 0
exclusive_sales = Decimal("0")
level1_bucket: dict[str, dict[str, dict[str, Decimal | int]]] = {
level1: {
"activity": {"qty": 0, "sales": Decimal("0")},
"ess": {"qty": 0, "sales": Decimal("0")},
"non_activity": {"qty": 0, "sales": Decimal("0")},
"ttl": {"qty": 0, "sales": Decimal("0")},
}
for level1 in LEVEL1_ORDER
}
current_table3_by_category: dict[str, dict[str, Decimal | int]] = {
name: {"qty": 0, "sales": Decimal("0")} for name in TABLE3_CATEGORY_ORDER
}
by_memo: dict[str, dict[str, Any]] = {}
top_rows: list[dict[str, Any]] = []
current_article_to_cn_category: dict[str, str] = {}
for row in shop_rows:
qty = to_int(row.get("qty"))
sales = to_decimal(row.get("sales"))
memo = (row.get("memo") or "").strip()
single = (row.get("single") or "").strip()
level1 = (row.get("level1") or "").strip().upper()
level3 = (row.get("level3") or "").strip()
article = (row.get("article") or "").strip()
total_qty += qty
total_sales += sales
bucket = by_memo.setdefault(
memo or "<EMPTY>",
{"rows": 0, "qty": 0, "sales": Decimal("0")},
)
bucket["rows"] += 1
bucket["qty"] += qty
bucket["sales"] += sales
if level1 in level1_bucket:
level1_bucket[level1]["ttl"]["qty"] += qty
level1_bucket[level1]["ttl"]["sales"] += sales
if memo in ACTIVITY_MEMOS:
activity_qty += qty
activity_sales += sales
if level1 in level1_bucket:
level1_bucket[level1]["activity"]["qty"] += qty
level1_bucket[level1]["activity"]["sales"] += sales
elif memo == ESS_MEMO:
ess_qty += qty
ess_sales += sales
if level1 in level1_bucket:
level1_bucket[level1]["ess"]["qty"] += qty
level1_bucket[level1]["ess"]["sales"] += sales
elif memo in NON_ACTIVITY_ONLY_MEMOS:
non_activity_only_qty += qty
non_activity_only_sales += sales
if level1 in level1_bucket:
level1_bucket[level1]["non_activity"]["qty"] += qty
level1_bucket[level1]["non_activity"]["sales"] += sales
if single == "Y":
exclusive_qty += qty
exclusive_sales += sales
if level3 in current_table3_by_category:
current_table3_by_category[level3]["qty"] += qty
current_table3_by_category[level3]["sales"] += sales
if article and level3:
current_article_to_cn_category[article] = level3
top_rows.append(
{
"article": article,
"memo": memo,
"single": single,
"qty": qty,
"sales": float(sales),
"level1": level1,
"level3": level3,
"activitybegin": row.get("activitybegin"),
"activityend": row.get("activityend"),
}
)
top_rows.sort(key=lambda item: item["sales"], reverse=True)
top_rows = top_rows[:20]
top10_rows = top_rows[:10]
asset_dir = workdir / "assets" / "campaign-s11"
top10_image_path = asset_dir / f"s11-top10-{period.ym}.png"
top10_image_cache_dir = asset_dir / "top10-images"
render_top10_panel_image(top10_rows, top10_image_path, image_cache_dir=top10_image_cache_dir)
qty_lfl_chart_image_path = asset_dir / f"s11-qty-lfl-{period.ym}.png"
sales_lfl_chart_image_path = asset_dir / f"s11-sales-lfl-{period.ym}.png"
summary_cur_sales = to_decimal(summary_cur.get("sales"))
summary_cur_uv = to_decimal(summary_cur.get("uv"))
summary_cur_tran = to_decimal(summary_cur.get("tran"))
summary_cur_qty = to_decimal(summary_cur.get("qty"))
summary_prev_sales = to_decimal(summary_prev.get("sales"))
summary_prev_uv = to_decimal(summary_prev.get("uv"))
summary_prev_tran = to_decimal(summary_prev.get("tran"))
summary_prev_qty = to_decimal(summary_prev.get("qty"))
cur_con = Decimal("0") if summary_cur_uv == 0 else summary_cur_tran / summary_cur_uv
prev_con = Decimal("0") if summary_prev_uv == 0 else summary_prev_tran / summary_prev_uv
cur_atv = Decimal("0") if summary_cur_tran == 0 else summary_cur_sales / summary_cur_tran
prev_atv = Decimal("0") if summary_prev_tran == 0 else summary_prev_sales / summary_prev_tran
# 文案中的“活动款及非活动款”口径延续现状：活动 vs (ESS + 非活动款)。
non_activity_for_ratio_qty = max(total_qty - activity_qty, 0)
non_activity_for_ratio_sales = total_sales - activity_sales
activity_ratio_qty = ratio_string(activity_qty, non_activity_for_ratio_qty)
activity_ratio_sales = ratio_string(
round_half_up(activity_sales),
round_half_up(non_activity_for_ratio_sales),
)
exclusive_ratio_qty = ratio_string(exclusive_qty, max(total_qty - exclusive_qty, 0))
sales_10k = int(total_sales / Decimal("10000"))
# 先构建同比期三级分类：优先 compare_ym 的 oms_shop_report；没有就走明细映射兜底。
compare_table3_by_category: dict[str, dict[str, Decimal | int]] = {
name: {"qty": 0, "sales": Decimal("0")} for name in TABLE3_CATEGORY_ORDER
}
table3_compare_source = "oms_shop_report"
if compare_shop_rows:
for row in compare_shop_rows:
category = (row.get("level3") or "").strip()
if category not in compare_table3_by_category:
continue
compare_table3_by_category[category]["qty"] += to_int(row.get("qty"))
compare_table3_by_category[category]["sales"] += to_decimal(row.get("sales"))
else:
table3_compare_source = "oms_daily_report_detail_positive_with_mapping"
# 当前期明细 article -> (level3, level4) 取销售最高组合作为映射锚点。
current_detail_article_feature: dict[str, tuple[str, str, Decimal]] = {}
for row in detail_cur_positive:
article = (row.get("articlecode") or "").strip()
level3 = (row.get("level3") or "").strip()
level4 = (row.get("level4") or "").strip()
sales = to_decimal(row.get("sales"))
if not article:
continue
previous = current_detail_article_feature.get(article)
if previous is None or sales > previous[2]:
current_detail_article_feature[article] = (level3, level4, sales)
# 构建 level3+level4 以及 level3 的票选映射，限制到表3可展示的中文类目。
combo_votes: dict[tuple[str, str], dict[str, Decimal]] = {}
level3_votes: dict[str, dict[str, Decimal]] = {}
for article, category in current_article_to_cn_category.items():
if category not in TABLE3_CATEGORY_SET:
continue
feature = current_detail_article_feature.get(article)
if feature is None:
continue
level3, level4, sales = feature
combo_votes.setdefault((level3, level4), {}).setdefault(category, Decimal("0"))
combo_votes[(level3, level4)][category] += sales
level3_votes.setdefault(level3, {}).setdefault(category, Decimal("0"))
level3_votes[level3][category] += sales
combo_to_cn: dict[tuple[str, str], str] = {}
for key, vote in combo_votes.items():
category = max(vote.items(), key=lambda item: item[1])[0]
combo_to_cn[key] = category
level3_to_cn: dict[str, str] = {}
for level3, vote in level3_votes.items():
category = max(vote.items(), key=lambda item: item[1])[0]
level3_to_cn[level3] = category
for row in detail_prev_positive:
article = (row.get("articlecode") or "").strip()
level3 = (row.get("level3") or "").strip()
level4 = (row.get("level4") or "").strip()
qty = to_int(row.get("qty"))
sales = to_decimal(row.get("sales"))
category = choose_cn_category_for_detail(
articlecode=article,
level3=level3,
level4=level4,
article_to_cn=current_article_to_cn_category,
combo_to_cn=combo_to_cn,
level3_to_cn=level3_to_cn,
)
compare_table3_by_category[category]["qty"] += qty
compare_table3_by_category[category]["sales"] += sales
# 图表与文案都依赖 bags/shoes 的同比。
current_bags_qty = sum(
to_int(current_table3_by_category[name]["qty"]) for name in TABLE3_BAGS_CATEGORY_SET if name in current_table3_by_category
)
current_shoes_qty = sum(
to_int(current_table3_by_category[name]["qty"]) for name in TABLE3_SHOES_CATEGORY_SET if name in current_table3_by_category
)
compare_bags_qty = sum(
to_int(compare_table3_by_category[name]["qty"]) for name in TABLE3_BAGS_CATEGORY_SET if name in compare_table3_by_category
)
compare_shoes_qty = sum(
to_int(compare_table3_by_category[name]["qty"]) for name in TABLE3_SHOES_CATEGORY_SET if name in compare_table3_by_category
)
current_bags_sales = sum(
to_decimal(current_table3_by_category[name]["sales"]) for name in TABLE3_BAGS_CATEGORY_SET if name in current_table3_by_category
)
current_shoes_sales = sum(
to_decimal(current_table3_by_category[name]["sales"]) for name in TABLE3_SHOES_CATEGORY_SET if name in current_table3_by_category
)
compare_bags_sales = sum(
to_decimal(compare_table3_by_category[name]["sales"]) for name in TABLE3_BAGS_CATEGORY_SET if name in compare_table3_by_category
)
compare_shoes_sales = sum(
to_decimal(compare_table3_by_category[name]["sales"]) for name in TABLE3_SHOES_CATEGORY_SET if name in compare_table3_by_category
)
shoes_sales_lfl_pct = round_half_up(safe_ratio(current_shoes_sales, compare_shoes_sales) * Decimal("100"))
sales_lfl_pct = round_half_up(safe_ratio(summary_cur_sales, summary_prev_sales) * Decimal("100"))
atv_lfl_pct = round_half_up(safe_ratio(cur_atv, prev_atv) * Decimal("100"))
strict_expected = (
period.report_year == 2026
and period.month_num == 1
and not args.disable_strict_january_check
)
strict_check = {
"enabled": strict_expected,
"passed": True,
"errors": [],
}
if strict_expected:
# 一月模板对齐要求“硬失败”：关键口径不一致直接中断，不继续渲染。
if sales_10k != JAN_TEMPLATE_EXPECTED["sales_10k"]:
strict_check["passed"] = False
strict_check["errors"].append(
f"sales_10k expected {JAN_TEMPLATE_EXPECTED['sales_10k']} but got {sales_10k}"
)
if activity_ratio_qty != JAN_TEMPLATE_EXPECTED["activity_ratio_qty"]:
strict_check["passed"] = False
strict_check["errors"].append(
"activity_ratio_qty expected "
f"{JAN_TEMPLATE_EXPECTED['activity_ratio_qty']} but got {activity_ratio_qty}"
)
if not strict_check["passed"]:
raise SystemExit(
"S11 一月模板强校验未通过: " + "; ".join(strict_check["errors"])
)
# S11 主叙述文本按模板句式替换关键数字，保持页面视觉风格不变。
narrative_template = load_s11_narrative_template(workdir)
narrative_text = rewrite_s11_narrative(
narrative_template,
sales_10k=sales_10k,
sales_lfl_pct=sales_lfl_pct,
atv_lfl_pct=atv_lfl_pct,
activity_ratio_qty=activity_ratio_qty,
shoes_sales_lfl_pct=shoes_sales_lfl_pct,
)
# 表1（Summary KPI）
table1_rows = [
[
campaign_window.label,
"SALES",
"UV",
"CON%",
"TRAN",
"ATV",
"QTY",
],
[
f"Y-{period.report_year % 100:02d}",
format_sales_2_with_comma(summary_cur_sales),
format_int_with_comma(round_half_up(summary_cur_uv)),
f"{(cur_con * Decimal('100')).quantize(Decimal('0.01'))}%",
format_int_with_comma(round_half_up(summary_cur_tran)),
format_plain_int(round_half_up(cur_atv)),
format_int_with_comma(round_half_up(summary_cur_qty)),
],
[
f"Y-{period.compare_year % 100:02d}",
format_sales_2_with_comma(summary_prev_sales),
format_int_with_comma(round_half_up(summary_prev_uv)),
f"{(prev_con * Decimal('100')).quantize(Decimal('0.01'))}%",
format_int_with_comma(round_half_up(summary_prev_tran)),
format_plain_int(round_half_up(prev_atv)),
format_int_with_comma(round_half_up(summary_prev_qty)),
],
[
"LFL",
ratio_percent_text(summary_cur_sales, summary_prev_sales),
ratio_percent_text(summary_cur_uv, summary_prev_uv),
ratio_percent_text(cur_con, prev_con),
ratio_percent_text(summary_cur_tran, summary_prev_tran),
ratio_percent_text(cur_atv, prev_atv),
ratio_percent_text(summary_cur_qty, summary_prev_qty),
],
]
# 表2（Type split）
table2_rows = [
[campaign_window.label, "Type", "qty", "qty%", "sales", "sales%"],
]
show_ess_rows = ess_qty > 0 or ess_sales > 0
for level1 in LEVEL1_ORDER:
level_data = level1_bucket[level1]
activity_qty_val = int(level_data["activity"]["qty"])
activity_sales_val = to_decimal(level_data["activity"]["sales"])
ess_qty_val = int(level_data["ess"]["qty"])
ess_sales_val = to_decimal(level_data["ess"]["sales"])
non_qty_val = int(level_data["non_activity"]["qty"])
non_sales_val = to_decimal(level_data["non_activity"]["sales"])
ttl_qty_val = int(level_data["ttl"]["qty"])
ttl_sales_val = to_decimal(level_data["ttl"]["sales"])
table2_rows.append(
[
level1,
"活动款",
format_int_with_comma(activity_qty_val),
pct_of_total_text(Decimal(activity_qty_val), Decimal(total_qty)),
format_sales_int_with_comma(activity_sales_val),
pct_of_total_text(activity_sales_val, total_sales),
]
)
if show_ess_rows:
table2_rows.append(
[
level1,
"ESS",
format_int_with_comma(ess_qty_val),
pct_of_total_text(Decimal(ess_qty_val), Decimal(total_qty)),
format_sales_int_with_comma(ess_sales_val),
pct_of_total_text(ess_sales_val, total_sales),
]
)
else:
table2_rows.append(["", "", "", "", "", ""])
table2_rows.append(
[
level1,
"非活动款",
format_int_with_comma(non_qty_val),
pct_of_total_text(Decimal(non_qty_val), Decimal(total_qty)),
format_sales_int_with_comma(non_sales_val),
pct_of_total_text(non_sales_val, total_sales),
]
)
table2_rows.append(
[
level1,
"TTL",
format_int_with_comma(ttl_qty_val),
pct_of_total_text(Decimal(ttl_qty_val), Decimal(total_qty)),
format_sales_int_with_comma(ttl_sales_val),
pct_of_total_text(ttl_sales_val, total_sales),
]
)
table2_rows.append(
[
"Grandtotal ",
"Grandtotal ",
format_int_with_comma(total_qty),
"100%",
format_sales_int_with_comma(total_sales),
"100%",
]
)
table2_rows.append(
[
"活动款",
"活动款",
format_plain_int(activity_qty),
pct_of_total_text(Decimal(activity_qty), Decimal(total_qty)),
format_sales_2_plain(activity_sales),
pct_of_total_text(activity_sales, total_sales),
]
)
if show_ess_rows:
table2_rows.append(
[
"ESS",
"ESS",
format_plain_int(ess_qty),
pct_of_total_text(Decimal(ess_qty), Decimal(total_qty)),
format_sales_2_plain(ess_sales),
pct_of_total_text(ess_sales, total_sales),
]
)
else:
table2_rows.append(["", "", "", "", "", ""])
table2_rows.append(
[
"非活动款",
"非活动款",
format_plain_int(non_activity_only_qty),
pct_of_total_text(Decimal(non_activity_only_qty), Decimal(total_qty)),
format_sales_2_plain(non_activity_only_sales),
pct_of_total_text(non_activity_only_sales, total_sales),
]
)
table2_rows.append(
[
"独家款",
"独家款",
format_plain_int(exclusive_qty),
pct_of_total_text(Decimal(exclusive_qty), Decimal(total_qty)),
format_sales_2_plain(exclusive_sales),
pct_of_total_text(exclusive_sales, total_sales),
]
)
# 表3（三级分类）
table3_rows = [
["三级分类", "Y-26 QTY", "Y-25 QTY", "QTY LFL", "Y-26 SALES", "Y-25 SALES", "SALES LFL"],
]
for category in TABLE3_CATEGORY_ORDER:
current_qty = Decimal(to_int(current_table3_by_category[category]["qty"]))
current_sales = to_decimal(current_table3_by_category[category]["sales"])
compare_qty = Decimal(to_int(compare_table3_by_category[category]["qty"]))
compare_sales = to_decimal(compare_table3_by_category[category]["sales"])
table3_rows.append(
[
category,
format_plain_int(round_half_up(current_qty)),
format_plain_int(round_half_up(compare_qty)),
ratio_percent_text(current_qty, compare_qty),
format_sales_trim_plain(current_sales),
format_sales_trim_plain(compare_sales),
ratio_percent_text(current_sales, compare_sales),
]
)
table3_rows.append(
[
"TTL",
format_plain_int(round_half_up(summary_cur_qty)),
format_plain_int(round_half_up(summary_prev_qty)),
ratio_percent_text(summary_cur_qty, summary_prev_qty),
format_sales_trim_plain(summary_cur_sales),
format_sales_trim_plain(summary_prev_sales),
ratio_percent_text(summary_cur_sales, summary_prev_sales),
]
)
qty_chart_series = [
{
"index": 1,
"name": "",
"x_values": ["BAGS", "SHOES"],
"values": [current_bags_qty, current_shoes_qty],
},
{
"index": 2,
"name": "",
"x_values": ["BAGS", "SHOES"],
"values": [compare_bags_qty, compare_shoes_qty],
},
{
"index": 3,
"name": "",
"values": [
float(safe_ratio(Decimal(current_bags_qty), Decimal(compare_bags_qty))),
float(safe_ratio(Decimal(current_shoes_qty), Decimal(compare_shoes_qty))),
],
},
]
sales_chart_series = [
{
"index": 1,
"name": "",
"x_values": ["BAGS", "SHOES"],
"values": [float(current_bags_sales), float(current_shoes_sales)],
},
{
"index": 2,
"name": "",
"x_values": ["BAGS", "SHOES"],
"values": [float(compare_bags_sales), float(compare_shoes_sales)],
},
{
"index": 3,
"name": "",
"values": [
float(safe_ratio(current_bags_sales, compare_bags_sales)),
float(safe_ratio(current_shoes_sales, compare_shoes_sales)),
],
},
]
# 图表在 COM 下直接写 Series 会受限，改为按数据库值生成图片替换。
render_lfl_chart_image(
output_path=qty_lfl_chart_image_path,
title="QTY LFL",
current_values=[float(current_bags_qty), float(current_shoes_qty)],
compare_values=[float(compare_bags_qty), float(compare_shoes_qty)],
lfl_values=[
float(safe_ratio(Decimal(current_bags_qty), Decimal(compare_bags_qty))),
float(safe_ratio(Decimal(current_shoes_qty), Decimal(compare_shoes_qty))),
],
)
render_lfl_chart_image(
output_path=sales_lfl_chart_image_path,
title="SALES LFL",
current_values=[float(current_bags_sales), float(current_shoes_sales)],
compare_values=[float(compare_bags_sales), float(compare_shoes_sales)],
lfl_values=[
float(safe_ratio(current_bags_sales, compare_bags_sales)),
float(safe_ratio(current_shoes_sales, compare_shoes_sales)),
],
)
operations = {
"replace_text": [
{
"slide": S11_NARRATIVE_SHAPE["slide"],
"shape_id": S11_NARRATIVE_SHAPE["shape_id"],
"shape_name": S11_NARRATIVE_SHAPE["shape_name"],
"new_text": narrative_text,
}
],
"replace_tables": [
{
"slide": S11_TABLE1_SHAPE["slide"],
"shape_id": S11_TABLE1_SHAPE["shape_id"],
"shape_name": S11_TABLE1_SHAPE["shape_name"],
"cells": build_table_cells(table1_rows),
},
{
"slide": S11_TABLE2_SHAPE["slide"],
"shape_id": S11_TABLE2_SHAPE["shape_id"],
"shape_name": S11_TABLE2_SHAPE["shape_name"],
"cells": build_table_cells(table2_rows),
},
{
"slide": S11_TABLE3_SHAPE["slide"],
"shape_id": S11_TABLE3_SHAPE["shape_id"],
"shape_name": S11_TABLE3_SHAPE["shape_name"],
"cells": build_table_cells(table3_rows),
},
],
"replace_charts": [],
"replace_images": [
{
"slide": S11_CHART_QTY_SHAPE["slide"],
"shape_id": S11_CHART_QTY_SHAPE["shape_id"],
"shape_name": S11_CHART_QTY_SHAPE["shape_name"],
"image_path": str(qty_lfl_chart_image_path),
},
{
"slide": S11_CHART_SALES_SHAPE["slide"],
"shape_id": S11_CHART_SALES_SHAPE["shape_id"],
"shape_name": S11_CHART_SALES_SHAPE["shape_name"],
"image_path": str(sales_lfl_chart_image_path),
},
{
"slide": S11_TOP10_SHAPE["slide"],
"shape_id": S11_TOP10_SHAPE["shape_id"],
"shape_name": S11_TOP10_SHAPE["shape_name"],
"image_path": str(top10_image_path),
}
],
}
summary = {
"shop_id": args.shop_id,
"month": period.ym,
"compare_month": period.compare_ym,
"campaign_window": {
"current_start": campaign_window.current_start.strftime("%Y-%m-%d"),
"current_end": campaign_window.current_end.strftime("%Y-%m-%d"),
"compare_start": campaign_window.compare_start.strftime("%Y-%m-%d"),
"compare_end": campaign_window.compare_end.strftime("%Y-%m-%d"),
"label": campaign_window.label,
},
"shop_report": {
"rows": len(shop_rows),
"compare_rows": len(compare_shop_rows),
"total_qty": total_qty,
"total_sales": float(total_sales),
"sales_10k_floor": sales_10k,
"activity_qty": activity_qty,
"activity_sales": float(activity_sales),
"non_activity_qty_for_ratio": non_activity_for_ratio_qty,
"non_activity_sales_for_ratio": float(non_activity_for_ratio_sales),
"non_activity_only_qty": non_activity_only_qty,
"non_activity_only_sales": float(non_activity_only_sales),
"ess_qty": ess_qty,
"ess_sales": float(ess_sales),
"show_ess_rows": show_ess_rows,
"exclusive_qty": exclusive_qty,
"exclusive_sales": float(exclusive_sales),
"activity_ratio_qty": activity_ratio_qty,
"activity_ratio_sales": activity_ratio_sales,
"exclusive_ratio_qty": exclusive_ratio_qty,
"by_memo": {
memo: {
"rows": bucket["rows"],
"qty": bucket["qty"],
"sales": float(bucket["sales"]),
}
for memo, bucket in sorted(by_memo.items(), key=lambda item: item[0])
},
"top10_generated_image": str(top10_image_path),
"top10_articles_by_sales": top10_rows,
"top20_articles_by_sales": top_rows,
},
"daily_report_summary": {
"current": {
"sales": float(summary_cur_sales),
"uv": float(summary_cur_uv),
"tran": round_half_up(summary_cur_tran),
"qty": round_half_up(summary_cur_qty),
"con_ratio": float(cur_con),
"atv": float(cur_atv),
},
"compare": {
"sales": float(summary_prev_sales),
"uv": float(summary_prev_uv),
"tran": round_half_up(summary_prev_tran),
"qty": round_half_up(summary_prev_qty),
"con_ratio": float(prev_con),
"atv": float(prev_atv),
},
"lfl_percent": {
"sales": sales_lfl_pct,
"uv": round_half_up(safe_ratio(summary_cur_uv, summary_prev_uv) * Decimal("100")),
"con": round_half_up(safe_ratio(cur_con, prev_con) * Decimal("100")),
"tran": round_half_up(safe_ratio(summary_cur_tran, summary_prev_tran) * Decimal("100")),
"atv": atv_lfl_pct,
"qty": round_half_up(safe_ratio(summary_cur_qty, summary_prev_qty) * Decimal("100")),
},
},
"table3_compare_source": table3_compare_source,
"table3_category_compare": {
name: {
"current_qty": to_int(current_table3_by_category[name]["qty"]),
"current_sales": float(to_decimal(current_table3_by_category[name]["sales"])),
"compare_qty": to_int(compare_table3_by_category[name]["qty"]),
"compare_sales": float(to_decimal(compare_table3_by_category[name]["sales"])),
}
for name in TABLE3_CATEGORY_ORDER
},
"chart_inputs": {
"bags_qty": {"current": current_bags_qty, "compare": compare_bags_qty},
"shoes_qty": {"current": current_shoes_qty, "compare": compare_shoes_qty},
"bags_sales": {"current": float(current_bags_sales), "compare": float(compare_bags_sales)},
"shoes_sales": {"current": float(current_shoes_sales), "compare": float(compare_shoes_sales)},
},
"strict_template_check": strict_check,
}
manifest_path = data_dir / f"s11-campaign-oms-shop-report.{period.ym}.json"
manifest_path.write_text(
json.dumps(
{
"source": {
"config_path": str(config_path),
"database": mysql_cfg["database"],
"tables": [
"oms_shop_report",
"oms_daily_report",
"oms_daily_report_detail",
],
"period": {
"month_label": period.month_label,
"year": period.report_year,
"compare_year": period.compare_year,
"ym": period.ym,
"compare_ym": period.compare_ym,
},
},
"summary": summary,
"narrative_preview": narrative_text,
},
ensure_ascii=False,
indent=2,
),
encoding="utf-8",
)
operations_path = workdir / "render-ops.campaign-s11.live.json"
operations_path.write_text(
json.dumps(operations, ensure_ascii=False, indent=2),
encoding="utf-8",
)
result = {
"operations_path": str(operations_path),
"manifest_path": str(manifest_path),
"summary": summary,
}
return result
def main() -> None:
args = parse_args()
result = run_sync(args)
print(json.dumps(result, ensure_ascii=False))
if __name__ == "__main__":
main()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment