XingQue/Django
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 新增 UFO 三角洲商品目录复制到龙先生的管理命令

Browse Source 
默认 dry-run，--apply 需双确认 club_id；只增不减，OSS 图物理复制。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
XingQue
2026-07-24 00:50:45 +08:00
parent 4b5130c824
commit e8e98e2970
1 changed files with 499 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

499
jituan/management/commands/copy_ufo_sanjiaozhou_catalog.py Normal file
View File

@@ -0,0 +1,499 @@
# -*- coding: utf-8 -*-
"""
将 UFO 俱乐部已上架的「三角洲」公共商品目录复制到龙先生俱乐部（只增不减）。
涉及表（只写目标俱乐部，不改源、不删任何行）：
- club_shangpin_leixing_config ClubShangpinLeixingConfig 上架类型
- shangpin_zhuanqu ShangpinZhuanqu 公共专区
- shangpin Shangpin 公共商品
- OSS copy_in_oss 图物理复制到新相对路径
不碰：
- shangpin_leixing（全局类型只复用，不新建）
- 店铺商品（dianpu 非空）
- 订单/统计/轮播/转盘活动表
- UFO 及其它俱乐部任何行
用法（服务器）:
cd /path/to/django
./venv/bin/python manage.py copy_ufo_sanjiaozhou_catalog --to-club=lxs
# 确认清单后：
./venv/bin/python manage.py copy_ufo_sanjiaozhou_catalog --to-club=lxs --apply --i-confirm-to-club=lxs
注意：龙先生 club_id 可能是 lxs 或 lsx，脚本不猜测，必须显式传 --to-club，
且 Club.name 必须包含「龙先生」。
"""
from __future__ import annotations
import time
import uuid
from django.core.management.base import BaseCommand, CommandError
from django.db import transaction
DEFAULT_FROM_CLUB = 'ufo'
DEFAULT_LEIXING_NAME = '三角洲'
NAME_HINT = '龙先生'
class Command(BaseCommand):
help = (
'复制 UFO「三角洲」公共上架商品目录到龙先生（默认 dry-run；'
'只 insert / 启用类型，不删不改源）'
)
def add_arguments(self, parser):
parser.add_argument(
'--from-club',
type=str,
default=DEFAULT_FROM_CLUB,
help=f'源俱乐部，默认 {DEFAULT_FROM_CLUB}',
)
parser.add_argument(
'--to-club',
type=str,
required=True,
help='目标俱乐部 club_id（必须显式指定，如 lxs 或 lsx）',
)
parser.add_argument(
'--leixing-name',
type=str,
default=DEFAULT_LEIXING_NAME,
help=f'全局类型名全名精确匹配，默认「{DEFAULT_LEIXING_NAME}」',
)
parser.add_argument(
'--leixing-id',
type=int,
default=0,
help='可选：直接指定全局 ShangpinLeixing.id（仍校验名称）',
)
parser.add_argument(
'--apply',
action='store_true',
help='真正写库 + OSS 复制（默认仅 dry-run）',
)
parser.add_argument(
'--i-confirm-to-club',
type=str,
default='',
help='与 --to-club 完全一致时才允许 --apply',
)
parser.add_argument(
'--list-clubs',
action='store_true',
help='只列出 Club 表（含名称含「龙先生」的候选），不复制',
)
def handle(self, *args, **options):
from jituan.models import Club, ClubShangpinLeixingConfig
from products.models import Shangpin, ShangpinLeixing, ShangpinZhuanqu
from utils.oss_utils import copy_in_oss
if options.get('list_clubs'):
self._list_clubs(Club)
return
from_club = (options.get('from_club') or DEFAULT_FROM_CLUB).strip()
to_club = (options.get('to_club') or '').strip()
leixing_name = (options.get('leixing_name') or DEFAULT_LEIXING_NAME).strip()
apply = bool(options.get('apply'))
confirm = (options.get('i_confirm_to_club') or '').strip()
if not to_club:
raise CommandError('必须传 --to-club=...')
self.stdout.write(self.style.NOTICE('—— Club 表候选（名称含「龙先生」）——'))
self._list_clubs(Club, only_hint=True)
src_club = Club.query.filter(club_id=from_club).first()
if not src_club:
raise CommandError(f'源俱乐部不存在: club_id={from_club!r}')
dst_club = Club.query.filter(club_id=to_club).first()
if not dst_club:
raise CommandError(
f'目标俱乐部不存在: club_id={to_club!r}。'
f'请先 --list-clubs 核对（常见：lxs 与 lsx 写反）。'
)
if NAME_HINT not in (dst_club.name or ''):
raise CommandError(
f'安全拦截：目标 club_id={to_club!r} 的 name={dst_club.name!r} '
f'不含「{NAME_HINT}」，拒绝执行。'
)
if from_club == to_club:
raise CommandError('源与目标 club_id 不能相同')
if apply:
if confirm != to_club:
raise CommandError(
'安全拦截：--apply 必须同时传 --i-confirm-to-club，且与 --to-club 完全一致。\n'
f' 例：--to-club={to_club} --apply --i-confirm-to-club={to_club}'
)
leixing = self._resolve_leixing(
ShangpinLeixing,
leixing_id=int(options.get('leixing_id') or 0),
leixing_name=leixing_name,
)
if leixing.shenhezhuangtai == 2:
raise CommandError(
f'类型 id={leixing.id}「{leixing.jieshao}」是审核专用，拒绝挂载'
)
# 源：公共专区 + 上架商品（dianpu 为空 = 俱乐部公共目录）
src_zones = list(
ShangpinZhuanqu.query.filter(
club_id=from_club,
leixing_id=leixing.id,
shenhezhuangtai=1,
dianpu__isnull=True,
).order_by('paixu', 'id')
)
zone_ids = [z.id for z in src_zones]
src_products = []
if zone_ids:
src_products = list(
Shangpin.query.filter(
club_id=from_club,
leixing_id=leixing.id,
zhuanqu_id__in=zone_ids,
shenhezhuangtai=1,
shangjia_zhuangtai=True,
dianpu__isnull=True,
).order_by('paixu', 'id')
)
# 计划：Config
cfg_existing = ClubShangpinLeixingConfig.query.filter(
club_id=to_club, leixing_id=leixing.id,
).first()
if cfg_existing is None:
cfg_action = 'CREATE_ENABLE'
elif not cfg_existing.is_enabled:
cfg_action = 'ENABLE_EXISTING'
else:
cfg_action = 'SKIP_ALREADY_ENABLED'
# 计划：专区
zone_plans = []
for z in src_zones:
exist = ShangpinZhuanqu.query.filter(
club_id=to_club,
leixing_id=leixing.id,
mingzi=z.mingzi,
dianpu__isnull=True,
).first()
if exist:
zone_plans.append({
'source': z,
'action': 'SKIP',
'target_id': exist.id,
'new_tupian_url': exist.tupian_url or '',
})
else:
zone_plans.append({
'source': z,
'action': 'CREATE',
'target_id': None,
'new_tupian_url': None,
})
zone_by_src_id = {p['source'].id: p for p in zone_plans}
product_plans = []
for p in src_products:
zp = zone_by_src_id.get(p.zhuanqu_id)
if not zp:
product_plans.append({
'source': p,
'action': 'SKIP_NO_ZONE',
'reason': f'zhuanqu_id={p.zhuanqu_id} 不在本批三角洲专区',
})
continue
planned_zq_id = zp.get('target_id')
if planned_zq_id:
exist_p = Shangpin.query.filter(
club_id=to_club,
zhuanqu_id=planned_zq_id,
biaoqian=p.biaoqian,
dianpu__isnull=True,
).first()
if exist_p:
product_plans.append({
'source': p,
'action': 'SKIP',
'zone_plan': zp,
'target_zhuanqu_id': planned_zq_id,
})
continue
product_plans.append({
'source': p,
'action': 'CREATE',
'zone_plan': zp,
'target_zhuanqu_id': planned_zq_id,
})
# —— 打印确认清单 ——
mode = 'APPLY' if apply else 'DRY-RUN'
self.stdout.write(self.style.NOTICE(
f'\n===== {mode} =====\n'
f'源: {from_club} ({src_club.name}) → 目标: {to_club} ({dst_club.name})\n'
f'类型: jieshao={leixing.jieshao!r} id={leixing.id}（全局复用，不新建）\n'
f'ClubShangpinLeixingConfig: {cfg_action}\n'
f'源专区={len(src_zones)} 源上架商品={len(src_products)}\n'
))
self.stdout.write('—— 专区计划 ——')
for i, zp in enumerate(zone_plans, 1):
z = zp['source']
self.stdout.write(
f' [{i}] {zp["action"]} src_id={z.id} mingzi={z.mingzi!r} '
f'paixu={z.paixu} tupian={z.tupian_url or "-"}'
+ (f' → existing_id={zp["target_id"]}' if zp['target_id'] else '')
)
self.stdout.write('—— 商品计划 ——')
for i, pp in enumerate(product_plans, 1):
p = pp['source']
zname = ''
zp = pp.get('zone_plan')
if zp:
zname = zp['source'].mingzi
self.stdout.write(
f' [{i}] {pp["action"]} src_id={p.id} {p.biaoqian!r} '
f'¥{p.jiage} zone={zname!r} '
f'tupian={p.tupian_url or "-"} guize={p.guize_tupian or "-"}'
+ (f' reason={pp.get("reason")}' if pp.get('reason') else '')
)
stats_preview = {
'cfg': cfg_action,
'zones_create': sum(1 for x in zone_plans if x['action'] == 'CREATE'),
'zones_skip': sum(1 for x in zone_plans if x['action'] == 'SKIP'),
'products_create': sum(1 for x in product_plans if x['action'] == 'CREATE'),
'products_skip': sum(1 for x in product_plans if x['action'] == 'SKIP'),
'products_skip_no_zone': sum(1 for x in product_plans if x['action'] == 'SKIP_NO_ZONE'),
}
self.stdout.write(self.style.NOTICE(f'\n统计预览: {stats_preview}'))
if not apply:
self.stdout.write(self.style.WARNING(
'\n以上仅计划，未写库、未拷 OSS。\n'
'确认无误后执行：\n'
f' python manage.py copy_ufo_sanjiaozhou_catalog '
f'--to-club={to_club} --apply --i-confirm-to-club={to_club}\n'
))
return
# —— APPLY：先 OSS，再 DB 事务 ——
def dest_key(category: str, src_url: str) -> str:
ext = 'png'
raw = (src_url or '').split('?', 1)[0]
if '.' in raw.rsplit('/', 1)[-1]:
ext = raw.rsplit('.', 1)[-1].lower()[:8] or 'png'
ts = int(time.time() * 1000)
rand = uuid.uuid4().hex[:8]
return f'club/{to_club}/catalog/{category}/{ts}_{rand}.{ext}'
def copy_img(src_url: str, category: str) -> str:
if not src_url:
return ''
# 已是相对路径或 URL：物理复制到新 key
key = dest_key(category, src_url)
if not copy_in_oss(src_url, key):
raise CommandError(f'OSS 复制失败，已中止（未写库）: {src_url} → {key}')
return key
# 预复制专区/商品图（仅 CREATE）
oss_ok = 0
for zp in zone_plans:
if zp['action'] != 'CREATE':
continue
z = zp['source']
if z.tupian_url:
zp['new_tupian_url'] = copy_img(z.tupian_url, 'zhuanqu')
oss_ok += 1
else:
zp['new_tupian_url'] = ''
for pp in product_plans:
if pp['action'] != 'CREATE':
continue
p = pp['source']
pp['new_tupian_url'] = copy_img(p.tupian_url, 'shangpin') if p.tupian_url else ''
if p.tupian_url:
oss_ok += 1
if p.guize_tupian:
pp['new_guize_tupian'] = copy_img(p.guize_tupian, 'guize')
oss_ok += 1
else:
pp['new_guize_tupian'] = ''
stats = {
'cfg_created': 0,
'cfg_enabled': 0,
'zones_created': 0,
'zones_skipped': 0,
'products_created': 0,
'products_skipped': 0,
'products_skipped_no_zone': 0,
'oss_copied': oss_ok,
}
with transaction.atomic():
cfg, created = ClubShangpinLeixingConfig.query.get_or_create(
club_id=to_club,
leixing_id=leixing.id,
defaults={
'is_enabled': True,
'paixu': int(leixing.paixu or 0),
'tupian_url': '',
},
)
if created:
stats['cfg_created'] = 1
elif not cfg.is_enabled:
cfg.is_enabled = True
cfg.save(update_fields=['is_enabled', 'UpdateTime'])
stats['cfg_enabled'] = 1
# 专区 insert + 映射
zone_id_map = {} # src_id -> dst_id
for zp in zone_plans:
z = zp['source']
if zp['action'] == 'SKIP':
zone_id_map[z.id] = zp['target_id']
stats['zones_skipped'] += 1
continue
# 事务内再查一次，防并发重复
exist = ShangpinZhuanqu.query.filter(
club_id=to_club,
leixing_id=leixing.id,
mingzi=z.mingzi,
dianpu__isnull=True,
).first()
if exist:
zone_id_map[z.id] = exist.id
stats['zones_skipped'] += 1
continue
row = ShangpinZhuanqu.query.create(
mingzi=z.mingzi,
leixing_id=leixing.id,
club_id=to_club,
tupian_url=zp.get('new_tupian_url') or '',
shenhezhuangtai=1,
paixu=int(z.paixu or 0),
shangjia_zhuangtai=bool(getattr(z, 'shangjia_zhuangtai', True)),
fengjin_zhuangtai=bool(getattr(z, 'fengjin_zhuangtai', True)),
dianpu=None,
dianpu_leixing=None,
)
zone_id_map[z.id] = row.id
zp['target_id'] = row.id
stats['zones_created'] += 1
for pp in product_plans:
p = pp['source']
if pp['action'] == 'SKIP_NO_ZONE':
stats['products_skipped_no_zone'] += 1
continue
if pp['action'] == 'SKIP':
stats['products_skipped'] += 1
continue
dst_zq = zone_id_map.get(p.zhuanqu_id)
if not dst_zq:
raise CommandError(
f'商品「{p.biaoqian}」src_id={p.id} 无法映射 zhuanqu_id={p.zhuanqu_id}'
)
exist_p = Shangpin.query.filter(
club_id=to_club,
zhuanqu_id=dst_zq,
biaoqian=p.biaoqian,
dianpu__isnull=True,
).first()
if exist_p:
stats['products_skipped'] += 1
continue
Shangpin.query.create(
biaoqian=p.biaoqian,
jiage=p.jiage,
kucun=p.kucun,
leixing_id=leixing.id,
zhuanqu_id=dst_zq,
zhenshi_xiaoliang=0,
duiwai_xiaoliang=0,
jieshao=p.jieshao,
xiadan_xuzhi=p.xiadan_xuzhi,
guize_tupian=pp.get('new_guize_tupian') or '',
yaoqiuleixing=p.yaoqiuleixing,
huiyuan_id=p.huiyuan_id,
yongjin=p.yongjin,
kaioi_ewai_dashou_fencheng=bool(p.kaioi_ewai_dashou_fencheng),
ewai_dashou_fencheng=p.ewai_dashou_fencheng,
tupian_url=pp.get('new_tupian_url') or '',
paixu=int(p.paixu or 0),
shenhezhuangtai=1,
shangjia_zhuangtai=True,
fengjin_zhuangtai=bool(p.fengjin_zhuangtai),
shenhe_zhuangtai=bool(getattr(p, 'shenhe_zhuangtai', False)),
club_id=to_club,
shi_zhuanpan=bool(getattr(p, 'shi_zhuanpan', False)),
dianpu=None,
dianpu_leixing=None,
)
stats['products_created'] += 1
self.stdout.write(self.style.SUCCESS(f'\n写入完成: {stats}'))
self.stdout.write(
'说明：未复制 Zhuanpan*；全局 ShangpinLeixing 未改；源 UFO 数据未动。'
)
def _list_clubs(self, Club, only_hint=False):
qs = Club.query.order_by('club_id')
for c in qs:
name = c.name or ''
if only_hint and NAME_HINT not in name:
continue
mark = ' <<-- 龙先生候选' if NAME_HINT in name else ''
self.stdout.write(
f' club_id={c.club_id!r} name={name!r} status={c.status}{mark}'
)
if only_hint:
hits = [c for c in Club.query.order_by('club_id') if NAME_HINT in (c.name or '')]
if not hits:
self.stdout.write(self.style.WARNING(' （无名称含「龙先生」的俱乐部）'))
def _resolve_leixing(self, ShangpinLeixing, leixing_id: int, leixing_name: str):
if leixing_id:
row = ShangpinLeixing.query.filter(id=leixing_id).first()
if not row:
raise CommandError(f'找不到 ShangpinLeixing id={leixing_id}')
if (row.jieshao or '') != leixing_name:
raise CommandError(
f'安全拦截：--leixing-id={leixing_id} 的名称是「{row.jieshao}」，'
f'与要求的精确名「{leixing_name}」不一致。'
)
return row
matches = list(ShangpinLeixing.query.filter(jieshao=leixing_name))
if not matches:
# 列出相近提示
self.stdout.write('全局商品类型清单：')
for lx in ShangpinLeixing.query.order_by('id'):
self.stdout.write(f' id={lx.id} name={lx.jieshao!r}')
raise CommandError(
f'未找到全局类型全名精确匹配「{leixing_name}」。'
f'请改后台类型名或 --leixing-id=... 指定。'
)
if len(matches) > 1:
ids = [m.id for m in matches]
raise CommandError(
f'全局类型「{leixing_name}」匹配到多条 id={ids}，拒绝猜测。'
f'请用 --leixing-id=... 指定唯一一条。'
)
return matches[0]