feat: 增加 import_ufo_banners，仅向 ufo 灌入轮播与我的背景
写入 Lunbo：accept_order / accept_order_banner2 / boss_center；不动其它俱乐部。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
XingQue
@@ -20,7 +20,8 @@
|
||||
"user_center": 0,
|
||||
"zones": 12
|
||||
},
|
||||
"images_total": 132
|
||||
"images_total": 132,
|
||||
"boss_center": 1
|
||||
},
|
||||
"banners": [
|
||||
{
|
||||
@@ -4023,7 +4024,8 @@
|
||||
"thumb_local": "images/misc/home_cat_1.jpg"
|
||||
}
|
||||
],
|
||||
"raffle": null
|
||||
"raffle": null,
|
||||
"bg_img": "https://ufo.qn.17v.tech/uploads/20251001/Fls0cQjtfsV7IUXl4cwqO6BFvPyh.jpg"
|
||||
},
|
||||
"platform_rule_images": [
|
||||
{
|
||||
@@ -4032,5 +4034,13 @@
|
||||
"url": "https://ufo.qn.17v.tech/uploads/20250929/FvOszrybjo3aAGdOBfyZ2VNIofi5.jpg",
|
||||
"local": "images/rules/platform_look_game_1_1.jpg"
|
||||
}
|
||||
],
|
||||
"boss_center": [
|
||||
{
|
||||
"url": "https://ufo.qn.17v.tech/uploads/20251001/Fls0cQjtfsV7IUXl4cwqO6BFvPyh.jpg",
|
||||
"kind": "boss_center",
|
||||
"page_key_target": "boss_center",
|
||||
"source": "getUserCenterInfo.bg_img"
|
||||
}
|
||||
]
|
||||
}
|
||||
177
jituan/management/commands/import_ufo_banners.py
Normal file
177
jituan/management/commands/import_ufo_banners.py
Normal file
@@ -0,0 +1,177 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
仅导入 UFO 展示图到 Lunbo（不碰商品）。只写 club_id=ufo。
|
||||
|
||||
覆盖：
|
||||
- accept_order 首页主轮播
|
||||
- accept_order_banner2 首页第二横幅
|
||||
- boss_center 「我的」顶区背景（原站 bg_img）
|
||||
|
||||
默认 dry-run；--apply 才写库。只删/写 ufo 对应 page_key，不影响 xq/xzj。
|
||||
|
||||
cd /opt/1panel/apps/openresty/nginx/www/sites/43.142.166.152.443/django
|
||||
git pull origin main
|
||||
./venv/bin/python manage.py import_ufo_banners
|
||||
./venv/bin/python manage.py import_ufo_banners --apply
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import mimetypes
|
||||
import os
|
||||
import re
|
||||
import ssl
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
from django.core.management.base import BaseCommand, CommandError
|
||||
from django.db import transaction
|
||||
|
||||
TARGET_CLUB = 'ufo'
|
||||
DEFAULT_CATALOG = Path(__file__).resolve().parents[2] / 'data' / 'ufo_catalog.json'
|
||||
|
||||
# 原站「我的」顶区背景（getUserCenterInfo.bg_img）；catalog 里没有时用这个兜底
|
||||
BOSS_CENTER_FALLBACK = (
|
||||
'https://ufo.qn.17v.tech/uploads/20251001/Fls0cQjtfsV7IUXl4cwqO6BFvPyh.jpg'
|
||||
)
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = '导入 UFO 轮播 + 我的背景到 Lunbo（仅 club=ufo）'
|
||||
|
||||
def add_arguments(self, parser):
|
||||
parser.add_argument('--apply', action='store_true')
|
||||
parser.add_argument('--catalog', type=str, default='')
|
||||
parser.add_argument('--skip-oss', action='store_true')
|
||||
parser.add_argument('--club-id', type=str, default=TARGET_CLUB)
|
||||
|
||||
def handle(self, *args, **options):
|
||||
club_id = (options.get('club_id') or '').strip()
|
||||
if club_id != TARGET_CLUB:
|
||||
raise CommandError(f'只允许 club_id={TARGET_CLUB}')
|
||||
|
||||
catalog_path = Path(options['catalog']).expanduser() if options['catalog'] else DEFAULT_CATALOG
|
||||
if not catalog_path.is_file():
|
||||
raise CommandError(f'找不到 {catalog_path}')
|
||||
|
||||
data = json.loads(catalog_path.read_text(encoding='utf-8'))
|
||||
banners = list(data.get('banners') or [])
|
||||
banners2 = list(data.get('banners2') or [])
|
||||
boss = list(data.get('boss_center') or [])
|
||||
if not boss:
|
||||
# catalog 内嵌或 misc
|
||||
misc = data.get('misc') or {}
|
||||
bg = misc.get('bg_img') or BOSS_CENTER_FALLBACK
|
||||
boss = [{'url': bg, 'kind': 'boss_center'}]
|
||||
|
||||
apply = bool(options['apply'])
|
||||
skip_oss = bool(options['skip_oss'])
|
||||
|
||||
self.stdout.write(self.style.NOTICE(
|
||||
f'club={TARGET_CLUB} accept_order={len(banners)} '
|
||||
f'banner2={len(banners2)} boss_center={len(boss)} '
|
||||
f'mode={"APPLY" if apply else "DRY-RUN"}'
|
||||
))
|
||||
for b in banners:
|
||||
self.stdout.write(f' [主轮播] {b.get("url")}')
|
||||
for b in banners2:
|
||||
self.stdout.write(f' [第二横幅] {b.get("url")}')
|
||||
for b in boss:
|
||||
self.stdout.write(f' [我的背景] {b.get("url")}')
|
||||
|
||||
if not apply:
|
||||
self.stdout.write(self.style.WARNING(
|
||||
'dry-run。确认后：\n ./venv/bin/python manage.py import_ufo_banners --apply'
|
||||
))
|
||||
return
|
||||
|
||||
from config.models import Lunbo
|
||||
from jituan.models import Club
|
||||
from jituan.services.display_config import (
|
||||
LUNBO_PAGE_ACCEPT_BANNER2,
|
||||
LUNBO_PAGE_ACCEPT_ORDER,
|
||||
LUNBO_PAGE_BOSS_CENTER,
|
||||
lunbo_image_type_for_page,
|
||||
)
|
||||
|
||||
if not Club.query.filter(club_id=TARGET_CLUB).exists():
|
||||
raise CommandError('俱乐部 ufo 不存在')
|
||||
|
||||
stats = {'ok': 0, 'fail': 0}
|
||||
|
||||
def to_oss(url: str) -> str:
|
||||
if not url:
|
||||
return ''
|
||||
if skip_oss:
|
||||
return url
|
||||
if not str(url).startswith('http'):
|
||||
return url
|
||||
path = self._upload_remote(url)
|
||||
if not path:
|
||||
stats['fail'] += 1
|
||||
raise CommandError(f'OSS 上传失败: {url}')
|
||||
stats['ok'] += 1
|
||||
return path
|
||||
|
||||
def replace_page(page_key, items):
|
||||
image_type = lunbo_image_type_for_page(page_key)
|
||||
# 只删本俱乐部本 page_key
|
||||
deleted, _ = Lunbo.query.filter(
|
||||
club_id=TARGET_CLUB,
|
||||
page_key=page_key,
|
||||
ImageType=image_type,
|
||||
).delete()
|
||||
self.stdout.write(f' clear {page_key} deleted={deleted}')
|
||||
n = 0
|
||||
for b in items:
|
||||
url = to_oss(b.get('url') or b.get('path') or '')
|
||||
if not url:
|
||||
continue
|
||||
Lunbo.query.create(
|
||||
club_id=TARGET_CLUB,
|
||||
page_key=page_key,
|
||||
ImageURL=url,
|
||||
ImageType=image_type,
|
||||
)
|
||||
n += 1
|
||||
return n
|
||||
|
||||
with transaction.atomic():
|
||||
n1 = replace_page(LUNBO_PAGE_ACCEPT_ORDER, banners)
|
||||
n2 = replace_page(LUNBO_PAGE_ACCEPT_BANNER2, banners2)
|
||||
n3 = replace_page(LUNBO_PAGE_BOSS_CENTER, boss)
|
||||
|
||||
self.stdout.write(self.style.SUCCESS(
|
||||
f'完成仅 ufo：主轮播={n1} 第二横幅={n2} 我的背景={n3} '
|
||||
f'OSS成功={stats["ok"]} 失败={stats["fail"]}'
|
||||
))
|
||||
|
||||
@staticmethod
|
||||
def _upload_remote(url: str) -> str:
|
||||
from utils.oss_utils import upload_to_oss
|
||||
|
||||
try:
|
||||
ctx = ssl.create_default_context()
|
||||
req = Request(url, headers={'User-Agent': 'UFO-Import/1.0'})
|
||||
with urlopen(req, timeout=60, context=ctx) as resp:
|
||||
raw = resp.read()
|
||||
ctype = resp.headers.get('Content-Type') or ''
|
||||
except Exception:
|
||||
return ''
|
||||
|
||||
ext = mimetypes.guess_extension((ctype or '').split(';')[0].strip()) or ''
|
||||
if not ext or ext == '.jpe':
|
||||
m = re.search(r'\.(jpg|jpeg|png|gif|webp|bmp)(?:\?|$)', url, re.I)
|
||||
ext = ('.' + m.group(1).lower()) if m else '.jpg'
|
||||
if ext == '.jpeg':
|
||||
ext = '.jpg'
|
||||
|
||||
ts = datetime.now().strftime('%Y%m%d%H%M%S')
|
||||
rel = f"a_long/lunbo/{ts}_{uuid.uuid4().hex[:8]}{ext}"
|
||||
bio = BytesIO(raw)
|
||||
bio.name = os.path.basename(rel)
|
||||
ok = upload_to_oss(bio, rel)
|
||||
return rel if ok else ''
|
||||
Reference in New Issue
Block a user