buff3.0/新建文件夹/count_selection.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
导出选品数据到Excel的脚本
条件：在售数量>=100，利润率10%，智能选品助手：低
"""
import os
import sys
import configparser
from pathlib import Path
from decimal import Decimal
from typing import Dict, List, Any, Optional, Tuple

import pymysql
from pymysql.cursors import DictCursor
import openpyxl
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side


def load_config():
    """从 config.ini 加载配置"""
    conf_path = Path(__file__).resolve().parent / "config.ini"
    if not conf_path.exists():
        print(f"错误: 未找到配置文件 {conf_path}", file=sys.stderr)
        sys.exit(1)
    config = configparser.ConfigParser()
    config.read(conf_path, encoding="utf-8")
    return config


def get_db_connection(config):
    """创建数据库连接"""
    return pymysql.connect(
        host=config.get("database", "host"),
        port=config.getint("database", "port"),
        user=config.get("database", "user"),
        password=config.get("database", "password"),
        database=config.get("database", "database"),
        charset=config.get("database", "charset"),
        cursorclass=DictCursor,
    )


def get_candidate_ids(conn, config: dict) -> list:
    """获取候选饰品ID列表"""
    conditions = ["1=1"]
    params = []

    pmin = config.get("buffPriceMin")
    pmax = config.get("buffPriceMax")
    if pmin is not None and pmax is not None:
        conditions.append("`当前在售最低价格` >= %s AND `当前在售最低价格` <= %s")
        params.extend([float(pmin), float(pmax)])
    elif pmin is not None:
        conditions.append("`当前在售最低价格` >= %s")
        params.append(float(pmin))
    elif pmax is not None:
        conditions.append("`当前在售最低价格` <= %s")
        params.append(float(pmax))

    smin = config.get("buffSalesMin")
    smax = config.get("buffSalesMax")
    if smin is not None and smax is not None:
        conditions.append("`在售数量` >= %s AND `在售数量` <= %s")
        params.extend([int(smin), int(smax)])
    elif smin is not None:
        conditions.append("`在售数量` >= %s")
        params.append(int(smin))
    elif smax is not None:
        conditions.append("`在售数量` <= %s")
        params.append(int(smax))

    exclude_ids = config.get("excludeIds", [])
    if exclude_ids:
        placeholders = ", ".join(["%s"] * len(exclude_ids))
        conditions.append(f"`BUFF饰品ID` NOT IN ({placeholders})")
        params.extend([str(x) for x in exclude_ids])

    where_sql = " AND ".join(conditions)
    sql = f"SELECT DISTINCT `BUFF饰品ID` FROM buff_goods_filtered_view WHERE {where_sql}"

    with conn.cursor() as cursor:
        cursor.execute(sql, params)
        rows = cursor.fetchall()
        return [str(r["BUFF饰品ID"]) for r in rows if r["BUFF饰品ID"]]


def get_goods_data(conn, goods_id: str) -> Optional[Dict[str, Any]]:
    """获取饰品完整数据"""
    sql = """
        SELECT `当前在售最低价格`, `7天在售最低价格平均价`, `30天在售最低价格平均价`
        FROM buff_goods_filtered_view WHERE `BUFF饰品ID` = %s LIMIT 1
    """
    with conn.cursor() as cursor:
        cursor.execute(sql, (goods_id,))
        row = cursor.fetchone()
        if not row or row["当前在售最低价格"] is None:
            return None
        return {
            "lowest": Decimal(str(row["当前在售最低价格"])),
            "avg7": Decimal(str(row["7天在售最低价格平均价"])) if row["7天在售最低价格平均价"] is not None else None,
            "avg30": Decimal(str(row["30天在售最低价格平均价"])) if row["30天在售最低价格平均价"] is not None else None,
        }


def get_wear_ranges(conn, goods_id: str) -> List[Tuple[Decimal, Decimal, Decimal]]:
    """获取磨损区间列表：(最小磨损值, 最大磨损值, 区间底价)"""
    sql = """
        SELECT `最小磨损值`, `最大磨损值`, `区间底价`
        FROM buff_goods_filtered_view
        WHERE `BUFF饰品ID` = %s
        ORDER BY `最小磨损值` ASC
    """
    with conn.cursor() as cursor:
        cursor.execute(sql, (goods_id,))
        rows = cursor.fetchall()
        out = []
        for r in rows:
            if r["最小磨损值"] is not None and r["最大磨损值"] is not None and r["区间底价"] is not None:
                out.append((
                    Decimal(str(r["最小磨损值"])),
                    Decimal(str(r["最大磨损值"])),
                    Decimal(str(r["区间底价"]))
                ))
        return out


def normal_ranges(ranges: List[Tuple]) -> List[Tuple]:
    """正常磨损区间：区间底价随最小磨损值递增而单调不升"""
    if not ranges:
        return []
    out = [ranges[0]]
    for i in range(1, len(ranges)):
        if ranges[i][2] <= out[-1][2]:
            out.append(ranges[i])
        else:
            break
    return out


def qualified_ranges(normal: List[Tuple], compare_price: Decimal) -> List[Tuple]:
    """满足利润的区间：区间底价 > 比较价，一旦不满足则截断"""
    out = []
    for r in normal:
        if r[2] > compare_price:
            out.append(r)
        else:
            break
    return out


def check_profit(conn, goods_id: str, profit_rate: float) -> Tuple[bool, Optional[Dict]]:
    """
    检查饰品是否符合利润条件
    返回: (是否通过, 利润信息)
    """
    goods_data = get_goods_data(conn, goods_id)
    if not goods_data:
        return False, None

    lowest = goods_data["lowest"]
    if lowest <= 0:
        return False, None

    compare_price = lowest * (Decimal("1") + Decimal(str(profit_rate)))
    ranges = get_wear_ranges(conn, goods_id)
    normal = normal_ranges(ranges)
    qualified = qualified_ranges(normal, compare_price)

    if not qualified:
        return False, None

    # 获取第一个符合条件区间的信息
    first_range = qualified[0]
    min_wear = float(first_range[0])
    max_wear = float(first_range[1])
    range_base_price = float(first_range[2])

    return True, {
        "lowest_price": float(lowest),
        "compare_price": float(compare_price),
        "profit_margin": float(compare_price) - float(lowest),
        "min_wear": min_wear,
        "max_wear": max_wear,
        "range_base_price": range_base_price,
    }


def get_comprehensive_data(conn, goods_id: str) -> Optional[Dict]:
    """从 buff_goods_comprehensive_view 获取完整数据"""
    sql = f"SELECT * FROM buff_goods_comprehensive_view WHERE `BUFF饰品ID` = %s LIMIT 1"
    with conn.cursor() as cursor:
        cursor.execute(sql, (goods_id,))
        return cursor.fetchone()


def export_to_excel(data: List[Dict], output_path: str):
    """将数据导出到Excel"""
    wb = openpyxl.Workbook()
    ws = wb.active
    ws.title = "选品数据"

    # 样式定义
    header_font = Font(bold=True, size=11)
    header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
    header_font_white = Font(bold=True, size=11, color="FFFFFF")
    thin_border = Border(
        left=Side(style='thin'),
        right=Side(style='thin'),
        top=Side(style='thin'),
        bottom=Side(style='thin')
    )

    # 设置列宽（根据字段名）
    column_widths = {
        "BUFF饰品ID": 15,
        "饰品名称": 30,
        "武器类型": 12,
        "品质": 10,
        "外观": 20,
        "当前在售最低价格": 18,
        "在售数量": 12,
        "7天在售最低价格平均价": 22,
        "30天在售最低价格平均价": 22,
        "磨损区间底价": 15,
        "参考利润": 15,
        "利润率": 12,
    }

    # 写入表头
    headers = [
        "BUFF饰品ID", "饰品名称", "武器类型", "品质", "外观",
        "当前在售最低价格", "在售数量", "7天在售最低价格平均价", "30天在售最低价格平均价",
        "磨损区间底价", "参考利润", "利润率"
    ]

    for col, header in enumerate(headers, start=1):
        cell = ws.cell(row=1, column=col, value=header)
        cell.font = header_font_white
        cell.fill = header_fill
        cell.alignment = Alignment(horizontal='center', vertical='center')
        cell.border = thin_border
        # 设置列宽
        if header in column_widths:
            ws.column_dimensions[openpyxl.utils.get_column_letter(col)].width = column_widths[header]

    # 写入数据
    for row_idx, item in enumerate(data, start=2):
        ws.cell(row=row_idx, column=1, value=item.get("BUFF饰品ID")).border = thin_border
        ws.cell(row=row_idx, column=2, value=item.get("饰品名称")).border = thin_border
        ws.cell(row=row_idx, column=3, value=item.get("武器类型")).border = thin_border
        ws.cell(row=row_idx, column=4, value=item.get("品质")).border = thin_border
        ws.cell(row=row_idx, column=5, value=item.get("外观")).border = thin_border
        ws.cell(row=row_idx, column=6, value=item.get("当前在售最低价格")).border = thin_border
        ws.cell(row=row_idx, column=7, value=item.get("在售数量")).border = thin_border
        ws.cell(row=row_idx, column=8, value=item.get("7天在售最低价格平均价")).border = thin_border
        ws.cell(row=row_idx, column=9, value=item.get("30天在售最低价格平均价")).border = thin_border
        ws.cell(row=row_idx, column=10, value=item.get("磨损区间底价")).border = thin_border
        ws.cell(row=row_idx, column=11, value=item.get("参考利润")).border = thin_border
        ws.cell(row=row_idx, column=12, value=item.get("利润率")).border = thin_border

    wb.save(output_path)


def main():
    # 筛选条件
    config = {
        "buffPriceMin": 5,
        "buffPriceMax": 200,
        "buffSalesMin": 100,
        "buffSalesMax": None,
        "profitRate": 10,
        "excludeIds": [],
    }

    print("开始导出选品数据...")
    print(f"筛选条件：")
    print(f"  - BUFF价格区间: {config['buffPriceMin']} - {config['buffPriceMax']} 元")
    print(f"  - 在售数量: >={config['buffSalesMin']}")
    print(f"  - 利润率: {config['profitRate']}%")
    print(f"  - 智能选品: 低")
    print("=" * 60)

    db_config = load_config()
    conn = get_db_connection(db_config)

    try:
        # 获取候选饰品
        print("\n[1/4] 获取候选饰品...")
        candidate_ids = get_candidate_ids(conn, config)
        print(f"候选饰品数量: {len(candidate_ids)}")

        # 筛选符合利润条件的饰品
        print("\n[2/4] 筛选符合利润条件的饰品...")
        profit_rate = config["profitRate"] / 100.0
        qualified_ids = []
        profit_info_map = {}

        for i, goods_id in enumerate(candidate_ids, start=1):
            if i % 200 == 0:
                print(f"  已处理 {i}/{len(candidate_ids)}...")

            passed, info = check_profit(conn, goods_id, profit_rate)
            if passed:
                qualified_ids.append(goods_id)
                profit_info_map[goods_id] = info

        print(f"符合利润条件的饰品: {len(qualified_ids)}")

        # 获取完整数据
        print("\n[3/4] 获取完整数据...")
        export_data = []
        for i, goods_id in enumerate(qualified_ids, start=1):
            if i % 200 == 0:
                print(f"  已处理 {i}/{len(qualified_ids)}...")

            row = get_comprehensive_data(conn, goods_id)
            if row:
                profit_info = profit_info_map.get(goods_id, {})
                row_data = {
                    "BUFF饰品ID": row.get("BUFF饰品ID"),
                    "饰品名称": row.get("饰品名称"),
                    "武器类型": row.get("武器类型"),
                    "品质": row.get("品质"),
                    "外观": row.get("外观"),
                    "当前在售最低价格": row.get("当前在售最低价格"),
                    "在售数量": row.get("在售数量"),
                    "7天在售最低价格平均价": row.get("7天在售最低价格平均价"),
                    "30天在售最低价格平均价": row.get("30天在售最低价格平均价"),
                    "磨损区间底价": profit_info.get("range_base_price"),
                    "参考利润": profit_info.get("profit_margin"),
                    "利润率": config["profitRate"],
                }
                export_data.append(row_data)

        # 导出到Excel
        print("\n[4/4] 导出到Excel...")
        script_dir = Path(__file__).resolve().parent
        output_path = script_dir / f"选品数据_{config['buffSalesMin']}_{config['profitRate']}pct.xlsx"
        export_to_excel(export_data, str(output_path))

        print(f"\n导出完成！")
        print(f"文件路径: {output_path}")
        print(f"导出数据: {len(export_data)} 条")

    finally:
        conn.close()


if __name__ == "__main__":
    main()