shanxia2.0/backend/app/services/db/auto_generate_plans.py

"""地磅计划按月自动生成 — 迁移自 simple-dibang-xiaoshou/auto-create-plan/auto_create_monthly_plan.py"""
from __future__ import annotations

import logging
from datetime import date, datetime, timedelta

from sqlalchemy import and_, or_
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.db_models import DbPurchasePlan
from app.services.db.plan_helpers import default_execution_status, generate_plan_no

logger = logging.getLogger(__name__)


def month_range(dt: date) -> tuple[date, date]:
    """返回 dt 所在月份的首日、末日。"""
    first = dt.replace(day=1)
    if first.month == 12:
        next_month = first.replace(year=first.year + 1, month=1, day=1)
    else:
        next_month = first.replace(month=first.month + 1, day=1)
    last = next_month - timedelta(days=1)
    return first, last


def get_last_month_range(now: date) -> tuple[date, date]:
    """返回上个月的首日、末日。"""
    first_this, _ = month_range(now)
    last_prev = first_this - timedelta(days=1)
    return month_range(last_prev)


def run_auto_generate_monthly_plans(
    db: Session,
    *,
    force: bool = False,
    now: date | None = None,
) -> dict:
    """
    按月自动生成采购计划。

    - 默认仅在每月 1 日执行（force=True 时跳过该限制，供手动触发验收）
    - 遍历所有租户，按上月计划模板生成本月计划
    """
    today = now or date.today()

    if today.day != 1 and not force:
        logger.info("当前不是每月 1 日且未 force，跳过自动生成计划")
        return {
            "created_count": 0,
            "skipped_reason": "not_first_day_of_month",
            "current_month": today.strftime("%Y-%m"),
        }

    current_month = today.strftime("%Y-%m")
    this_month_first, this_month_last = month_range(today)
    last_month_first, last_month_last = get_last_month_range(today)

    logger.info(
        "开始执行自动生成计划任务，current_month=%s, last_month=[%s, %s], this_month=[%s, %s], force=%s",
        current_month,
        last_month_first,
        last_month_last,
        this_month_first,
        this_month_last,
        force,
    )

    plans = (
        db.query(DbPurchasePlan)
        .filter(
            and_(
                DbPurchasePlan.del_flag == 0,
                DbPurchasePlan.auto_generate_enabled == 1,
                or_(
                    DbPurchasePlan.auto_generate_last_month.is_(None),
                    DbPurchasePlan.auto_generate_last_month != current_month,
                ),
                or_(
                    DbPurchasePlan.auto_generate_end_month.is_(None),
                    DbPurchasePlan.auto_generate_end_month >= current_month,
                ),
                DbPurchasePlan.start_date == last_month_first,
                DbPurchasePlan.end_date == last_month_last,
            )
        )
        .all()
    )

    if not plans:
        logger.info("未找到需要自动生成下月计划的记录")
        return {
            "created_count": 0,
            "skipped_reason": None,
            "current_month": current_month,
        }

    logger.info("找到 %d 条可自动生成的上月计划", len(plans))
    created_count = 0

    for source in plans:
        new_plan = DbPurchasePlan(
            id=generate_uuid(),
            tenant_no=source.tenant_no,
            plan_no=generate_plan_no(),
            plan_date=this_month_first,
            start_date=this_month_first,
            end_date=this_month_last,
            business_type=source.business_type,
            product_id=source.product_id,
            product=source.product,
            specification=source.specification,
            planned_quantity=source.planned_quantity,
            vehicle_no=source.vehicle_no or "",
            customer_name=source.customer_name,
            remark=source.remark,
            execution_status=default_execution_status(source.business_type),
            executed_at=None,
            last_weighing_slip_id=None,
            executed_quantity=0,
            auto_generate_enabled=source.auto_generate_enabled,
            auto_generate_end_month=source.auto_generate_end_month,
            auto_generate_last_month=None,
            create_by=None,
            create_dept_id="",
        )
        db.add(new_plan)
        source.auto_generate_last_month = current_month
        created_count += 1

    db.commit()
    logger.info("本次自动生成完成，共生成 %d 条新计划", created_count)
    return {
        "created_count": created_count,
        "skipped_reason": None,
        "current_month": current_month,
    }