#!/usr/bin/env python3 # -*- coding: utf-8 -*- # 小商店 v4.0 —— 账户 + 订单历史 + 支付截图审核 + 国旗 import json, os, time, base64, urllib.request, urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer API_KEY = os.environ.get("HERO_API_KEY", "75991A808A1e95b606ed6142d027db66").strip() BOSS_KEY = os.environ.get("BOSS_KEY", "boss888") PORT = int(os.environ.get("PORT", "8888")) RATE = 7.2 MARKUP = float(os.environ.get("MARKUP", "3")) MANUAL_PRICE = float(os.environ.get("MANUAL_PRICE", "19.9")) # TG成品号（精品）售价 CONTACT = os.environ.get("CONTACT", "请扫码或添加店主微信付款，付款后请上传支付截图并点击[我已付款]") BASE = os.path.dirname(os.path.abspath(__file__)) STORE_FILE = os.path.join(BASE, "store.json") USERS_FILE = os.path.join(BASE, "users.json") QR_FILE = os.path.join(BASE, "收款码.png") API = "https://hero-sms.com/stubs/handler_api.php" REST = "https://hero-sms.com/api/v1" FLAGS = {"1": "🇺🇦", "10": "🇻🇳", "100": "🇰🇼", "101": "🇸🇻", "102": "🇱🇾", "103": "🇯🇲", "104": "🇹🇹", "105": "🇪🇨", "106": "🇸🇿", "107": "🇴🇲", "108": "🇧🇦", "109": "🇩🇴", "11": "🇰🇬", "110": "🇸🇾", "111": "🇶🇦", "112": "🇵🇦", "113": "🇨🇺", "114": "🇲🇷", "115": "🇸🇱", "116": "🇯🇴", "117": "🇵🇹", "118": "🇧🇧", "119": "🇧🇮", "120": "🇧🇯", "121": "🇧🇳", "122": "🇧🇸", "123": "🇧🇼", "124": "🇧🇿", "125": "🇨🇫", "126": "🇩🇲", "127": "🇬🇩", "128": "🇬🇪", "129": "🇬🇷", "13": "🇮🇱", "130": "🇬🇼", "131": "🇬🇾", "132": "🇮🇸", "133": "🇰🇲", "134": "🇰🇳", "135": "🇱🇷", "136": "🇱🇸", "137": "🇲🇼", "138": "🇳🇦", "139": "🇳🇪", "14": "🇭🇰", "140": "🇷🇼", "141": "🇸🇰", "142": "🇸🇷", "143": "🇹🇯", "144": "🇲🇨", "145": "🇧🇭", "146": "🇷🇪", "147": "🇿🇲", "148": "🇦🇲", "149": "🇸🇴", "15": "🇵🇱", "150": "🇨🇬", "151": "🇨🇱", "152": "🇧🇫", "153": "🇱🇧", "154": "🇬🇦", "155": "🇦🇱", "156": "🇺🇾", "157": "🇲🇺", "158": "🇧🇹", "159": "🇲🇻", "16": "🇬🇧", "160": "🇬🇵", "161": "🇹🇲", "162": "🇬🇫", "163": "🇫🇮", "164": "🇱🇨", "165": "🇱🇺", "166": "🇻🇨", "167": "🇬🇶", "168": "🇩🇯", "169": "🇦🇬", "17": "🇲🇬", "170": "🇰🇾", "171": "🇲🇪", "172": "🇩🇰", "173": "🇨🇭", "174": "🇳🇴", "175": "🇦🇺", "176": "🇪🇷", "177": "🇸🇸", "178": "🇸🇹", "179": "🇦🇼", "18": "🇨🇩", "180": "🇲🇸", "181": "🇦🇮", "182": "🇯🇵", "183": "🇲🇰", "184": "🇸🇨", "185": "🇳🇨", "186": "🇨🇻", "187": "🇺🇸", "188": "🇵🇸", "189": "🇫🇯", "19": "🇳🇬", "196": "🇸🇬", "198": "🇼🇸", "199": "🇲🇹", "2": "🇰🇿", "20": "🇲🇴", "200": "🇱🇮", "201": "🇬🇮", "203": "🇽🇰", "204": "🇳🇺", "21": "🇪🇬", "22": "🇮🇳", "23": "🇮🇪", "24": "🇰🇭", "25": "🇱🇦", "26": "🇭🇹", "27": "🇨🇮", "28": "🇬🇲", "29": "🇷🇸", "3": "🇨🇳", "30": "🇾🇪", "31": "🇿🇦", "32": "🇷🇴", "33": "🇨🇴", "34": "🇪🇪", "35": "🇦🇿", "36": "🇨🇦", "37": "🇲🇦", "38": "🇬🇭", "39": "🇦🇷", "4": "🇵🇭", "40": "🇺🇿", "41": "🇨🇲", "42": "🇹🇩", "43": "🇩🇪", "44": "🇱🇹", "45": "🇭🇷", "46": "🇸🇪", "47": "🇮🇶", "48": "🇳🇱", "49": "🇱🇻", "5": "🇲🇲", "50": "🇦🇹", "51": "🇧🇾", "52": "🇹🇭", "53": "🇸🇦", "54": "🇲🇽", "55": "🇹🇼", "56": "🇪🇸", "57": "🇮🇷", "58": "🇩🇿", "59": "🇸🇮", "6": "🇮🇩", "60": "🇧🇩", "61": "🇸🇳", "62": "🇹🇷", "63": "🇨🇿", "64": "🇱🇰", "65": "🇵🇪", "66": "🇵🇰", "67": "🇳🇿", "68": "🇬🇳", "69": "🇲🇱", "7": "🇲🇾", "70": "🇻🇪", "71": "🇪🇹", "72": "🇲🇳", "73": "🇧🇷", "74": "🇦🇫", "75": "🇺🇬", "76": "🇦🇴", "77": "🇨🇾", "78": "🇫🇷", "79": "🇵🇬", "8": "🇰🇪", "80": "🇲🇿", "81": "🇳🇵", "82": "🇧🇪", "83": "🇧🇬", "84": "🇭🇺", "85": "🇲🇩", "86": "🇮🇹", "87": "🇵🇾", "88": "🇭🇳", "89": "🇹🇳", "9": "🇹🇿", "90": "🇳🇮", "91": "🇹🇱", "92": "🇧🇴", "93": "🇨🇷", "94": "🇬🇹", "95": "🇦🇪", "96": "🇿🇼", "97": "🇵🇷", "98": "🇸🇩", "99": "🇹🇬"} _cache = {"t": 0, "countries": None, "services": None} _pcache = {"t": 0, "data": None} def preload(): try: cached_list() print(" ✅ 国家/服务清单预加载完成") except Exception as e: print(" ⚠️ 预加载失败:", e) def get(**p): p["api_key"] = API_KEY u = API + "?" + urllib.parse.urlencode(p) t = urllib.request.urlopen(u, timeout=15).read().decode() try: return json.loads(t) except Exception: return t def cached_list(): if time.time() - _cache["t"] > 600 or _cache["countries"] is None: try: _cache["countries"] = get(action="getCountries") _cache["services"] = get(action="getServicesList").get("services", []) _cache["t"] = time.time() except Exception: pass return _cache["countries"] or {}, _cache["services"] or [] def raw_name(cid): cs, _ = cached_list() c = cs.get(str(cid), {}) return c.get("chn") or c.get("eng") or f"国家{cid}" def cname_of(cid): cs, _ = cached_list() c = cs.get(str(cid), {}) return (FLAGS.get(str(cid), "") + (c.get("chn") or c.get("eng") or "")).strip() def store_load(): try: with open(STORE_FILE, encoding="utf-8") as f: d = json.load(f) except Exception: return {} now = time.time(); changed = False for k in list(d.keys()): o = d[k]; ts = o.get("ts", 0) if o["state"] == "wait_pay" and now - ts > 600: del d[k]; changed = True # 已付款(paid)订单不再自动作废，一直保留等待店主审核 if changed: store_save(d) return d def store_save(d): with open(STORE_FILE, "w", encoding="utf-8") as f: json.dump(d, f, ensure_ascii=False) def users_load(): try: with open(USERS_FILE, encoding="utf-8") as f: return json.load(f) except Exception: return {} def users_save(d): with open(USERS_FILE, "w", encoding="utf-8") as f: json.dump(d, f, ensure_ascii=False) def yuan(cost): v = float(cost) * RATE * MARKUP return round(max(1, int(v * 10 + 0.999)) / 10, 1) STATUS = {1: "待接单", 3: "已接单", 4: "等待短信", 5: "已发码", 6: "已完成", 8: "已取消"} # ---------------- 前端页面 ---------------- PAGE = """ 接码小铺

接码小铺

全球接码 · 自动交付

📢 服务说明

欢迎访问接码小铺。下单前请仔细核对商品说明、适用范围与交付方式，并确保提交的信息准确有效。数字商品通常会在付款确认后自动交付；如遇支付、订购或交付异常，请保留订单信息并通过站内公布的联系方式与我们联系。请合理、合规使用本站提供的商品与服务。

今日不再提示不再提示
🔍

📦 成品服务

店主自养精品成品号上架中，敬请期待。
上架后可在这里直接浏览和购买（TG 成品号等）。
如有急需请到【设置】联系客服咨询。
TG新号：付款确认后自动发货，20分钟未收到短信自动退款
""" PAGE = PAGE.replace("%MANUAL_PRICE%", str(MANUAL_PRICE)).replace("%CONTACT%", CONTACT) BOSS = """ 商店后台

🔐 商店后台

🛒 客户自助订单

加载中...

📋 我的活跃订单

加载中...
""" # ---------------- 服务端 ---------------- class H(BaseHTTPRequestHandler): def log_message(self, fmt, *a): pass def _send(self, code, body, ctype="text/html; charset=utf-8"): b = body.encode() self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(b))) self.end_headers() self.wfile.write(b) def _json(self, obj): self._send(200, json.dumps(obj, ensure_ascii=False), "application/json; charset=utf-8") def _body(self): return json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))) or b"{}") def do_GET(self): u = urllib.parse.urlparse(self.path) q = urllib.parse.parse_qs(u.query) if u.path == "/": return self._send(200, PAGE) if u.path == "/api/code": aid = (q.get("id") or [""])[0].strip() so = None if not aid: return self._json({"error": "请输入订单号"}) # 支持店铺订单号 SO...：转到对应平台单号或显示店铺状态 if aid.upper().startswith("SO"): o = store_load().get(aid.upper()) if not o: return self._json({"error": "店铺订单不存在（未付款订单超10分钟自动清除）"}) so = o if o.get("myid"): aid = str(o["myid"]) elif o["state"] == "wait_pay": return self._json({"soState": "wait_pay", "left": max(0, int(600 - (time.time() - o.get("ts", 0))))}) elif o["state"] == "paid": return self._json({"soState": "paid", "left": max(0, int(180 - (time.time() - o.get("ts", 0))))}) elif o["state"] == "expired": return self._json({"error": "⏰ 该订单付款确认超时已作废，请重新下单"}) else: return self._json({"error": "该订单未发货（" + o["state"] + "），请联系店主"}) try: req = urllib.request.Request(REST + "/activations?ids=" + aid, headers={"Authorization": "ApiKey " + API_KEY}) items = json.loads(urllib.request.urlopen(req, timeout=15).read().decode()).get("data") or [] it = next((x for x in items if str(x.get("id")) == aid), None) if it: from datetime import datetime try: exp = datetime.fromisoformat(it["expiredAt"].replace("Z", "+00:00")) remain = max(0, int(exp.timestamp() - time.time())) except Exception: remain = 0 return self._json({"phone": it["phone"], "status": int(it["status"]), "code": (it.get("smsCode") or it.get("otpLast") or "").strip(), "remain": remain, "cname": cname_of(it.get("country")), "flag": FLAGS.get(str(it.get("country")), ""), "service": it.get("service", "")}) except Exception: pass try: r = get(action="getStatus", id=aid) if r == "STATUS_OK": return self._json({"phone": "", "status": 6, "code": "", "remain": 0}) if r == "STATUS_WAIT_CODE": return self._json({"phone": "", "status": 4, "code": "", "remain": 0}) if r == "STATUS_CANCEL": return self._json({"error": "该订单已取消退款，请重新下单"}) if r.startswith("STATUS_OK:"): return self._json({"phone": "", "status": 6, "code": r.split(":")[1], "remain": 0}) except Exception: pass # 平台查询失败时，用本地缓存的号码/验证码兜底 if so is not None and (so.get("code") or so.get("phone")): return self._json({"phone": so.get("phone", ""), "status": 6, "code": so.get("code", ""), "remain": 0, "cname": so.get("cname", ""), "flag": FLAGS.get(str(so.get("country")), ""), "service": so.get("service", "")}) return self._json({"error": "查不到这个订单，请核对 ID"}) if u.path == "/api/services": _, sv = cached_list() return self._json({"services": [{"code": s["code"], "name": s["name"], "alias": ALIAS.get(s["code"], "")} for s in sv]}) if u.path == "/api/prices": try: if _pcache["data"] and time.time() - _pcache["t"] < 120: return self._json({"prices": _pcache["data"]}) p = get(action="getPrices") out = {} for cid, svcs in p.items(): if not isinstance(svcs, dict): continue for svc, info in svcs.items(): if isinstance(info, dict) and info.get("count", 0) > 0: out.setdefault(svc, []).append( {"cid": int(cid), "name": raw_name(cid), "flag": FLAGS.get(str(cid), ""), "yuan": yuan(info.get("cost", 0)), "count": info.get("count")}) for rows in out.values(): rows.sort(key=lambda r: r["yuan"]) _pcache.update({"t": time.time(), "data": out}) return self._json({"prices": out}) except Exception as e: return self._json({"error": str(e)}) if u.path == "/api/neworder": b = self._body() st = store_load() oid = "SO" + str(int(time.time() * 10))[-9:] + str(len(st) % 10) if b.get("spec") == "manual": # TG成品号（精品）：手动发货，固定价 st[oid] = {"service": "tg成品", "country": None, "spec": "manual", "cname": "Telegram 成品号（精品）", "user": b.get("user", ""), "remark": b.get("remark", ""), "state": "wait_pay", "ts": time.time(), "created": time.strftime("%m-%d %H:%M"), "yuan": MANUAL_PRICE} store_save(st) return self._json({"id": oid, "yuan": MANUAL_PRICE, "cname": "Telegram 成品号（精品）", "service": "TG成品", "flag": "✈️", "contact": CONTACT, "hasQr": os.path.exists(QR_FILE)}) try: p = get(action="getPrices") cost = p[str(b["country"])][b["service"]]["cost"] except Exception: cost = 0 st[oid] = {"service": b.get("service"), "country": b.get("country"), "cname": b.get("cname", ""), "user": b.get("user", ""), "state": "wait_pay", "ts": time.time(), "created": time.strftime("%m-%d %H:%M"), "yuan": yuan(cost)} store_save(st) if True: return self._json({"id": oid, "yuan": yuan(cost), "cname": b.get("cname", ""), "service": b["service"], "flag": FLAGS.get(str(b["country"]), ""), "contact": CONTACT, "hasQr": os.path.exists(QR_FILE)}) if u.path == "/api/paid": b = self._body() st = store_load() o = st.get(b.get("id", "")) if not o: return self._json({"error": "订单不存在或已超时"}) if o["state"] in ("wait_pay", "paid"): o["state"] = "paid" o["ts"] = time.time() if b.get("img"): try: o["shot"] = "shot_" + b["id"] + ".png" with open(os.path.join(BASE, o["shot"]), "wb") as f: f.write(base64.b64decode(b["img"].split(",", 1)[1])) except Exception: pass store_save(st) return self._json({"ok": 1}) if u.path == "/api/myorder": aid = (q.get("id") or [""])[0].strip() o = store_load().get(aid) if not o: return self._json({"error": "订单不存在（未付款订单超10分钟自动清除）"}) d = {"state": o["state"], "id": aid, "cname": o.get("cname", ""), "service": o.get("service", ""), "flag": FLAGS.get(str(o.get("country")), ""), "yuan": o.get("yuan", "?"), "hasQr": os.path.exists(QR_FILE), "contact": CONTACT, "spec": o.get("spec", "auto"), "manual": 1 if o.get("spec") == "manual" else 0} if o["state"] == "wait_pay": d["left"] = max(0, int(600 - (time.time() - o.get("ts", 0)))) return self._json(d) if o["state"] == "paid": d["left"] = max(0, int(180 - (time.time() - o.get("ts", 0)))) return self._json(d) if o["state"] == "active" and o.get("myid"): try: req = urllib.request.Request(REST + "/activations?ids=" + str(o["myid"]), headers={"Authorization": "ApiKey " + API_KEY}) items = json.loads(urllib.request.urlopen(req, timeout=15).read().decode()).get("data") or [] it = next((x for x in items if str(x.get("id")) == str(o["myid"])), None) if it: from datetime import datetime try: exp = datetime.fromisoformat(it["expiredAt"].replace("Z", "+00:00")) d["remain"] = max(0, int(exp.timestamp() - time.time())) except Exception: d["remain"] = 0 d.update({"phone": it["phone"], "status": int(it["status"]), "code": (it.get("smsCode") or it.get("otpLast") or "").strip(), "cname": cname_of(it.get("country")), "service": it.get("service", "")}) try: _st = store_load() _o = _st.get(aid) if _o: if it["phone"]: _o["phone"] = str(it["phone"]) _cd = (it.get("smsCode") or it.get("otpLast") or "").strip() if _cd: _o["code"] = _cd store_save(_st) except Exception: pass except Exception: pass return self._json(d) if u.path == "/api/reg": b = self._body() us = users_load() a = (b.get("u") or "").strip() if not a or len(a) < 2: return self._json({"error": "用户名至少2个字符"}) if a in us: return self._json({"error": "该用户名已被注册，请换一个或直接登录"}) us[a] = {"p": b.get("p", ""), "created": time.strftime("%m-%d %H:%M")} users_save(us) return self._json({"ok": 1}) if u.path == "/api/login": b = self._body() us = users_load() a = (b.get("u") or "").strip() if a not in us: return self._json({"error": "用户不存在，请先注册"}) if us[a].get("p", "") != b.get("p", ""): return self._json({"error": "密码不对"}) return self._json({"ok": 1}) if u.path == "/api/myorders": b = self._body() us = users_load() a = (b.get("u") or "").strip() if a not in us or us[a].get("p", "") != b.get("p", ""): return self._json({"error": "请重新登录"}) st = store_load() ords = [{"id": k, "flag": FLAGS.get(str(v.get("country")), ""), "yuan": v.get("yuan", "?"), "cname": v.get("cname", ""), "state": v["state"], "code": v.get("code", ""), "myid": v.get("myid", ""), "phone": v.get("phone", ""), "service": v.get("service", ""), "manual": 1 if v.get("spec") == "manual" else 0} for k, v in st.items() if v.get("user") == a] ords.sort(key=lambda x: x["id"], reverse=True) return self._json({"orders": ords}) if u.path == "/qrcode.png" and os.path.exists(QR_FILE): with open(QR_FILE, "rb") as f: b = f.read() self.send_response(200) self.send_header("Content-Type", "image/png") self.send_header("Content-Length", str(len(b))) self.end_headers() self.wfile.write(b) return if u.path == "/boss": return self._send(200, BOSS) if u.path == "/boss/api/balance": if (q.get("key") or [""])[0] != BOSS_KEY: return self._json({"auth": 0}) try: return self._json({"balance": get(action="getBalance").split(":")[1]}) except Exception as e: return self._json({"error": str(e)}) if u.path == "/boss/api/services": if (q.get("key") or [""])[0] != BOSS_KEY: return self._json({"auth": 0}) _, sv = cached_list() return self._json({"services": [{"code": s["code"], "name": s["name"]} for s in sv], "alias": {s["code"]: ALIAS.get(s["code"], "") for s in sv}}) if u.path == "/boss/api/prices": if (q.get("key") or [""])[0] != BOSS_KEY: return self._json({"auth": 0}) try: p = get(action="getPrices") out = {} for cid, svcs in p.items(): if not isinstance(svcs, dict): continue for svc, info in svcs.items(): if isinstance(info, dict) and info.get("count", 0) > 0: out.setdefault(svc, []).append( {"cid": int(cid), "name": raw_name(cid), "flag": FLAGS.get(str(cid), ""), "cost": float(info.get("cost", 0)), "count": info.get("count")}) for rows in out.values(): rows.sort(key=lambda r: r["cost"]) return self._json({"prices": out}) except Exception as e: return self._json({"error": str(e)}) if u.path == "/boss/api/orders": if (q.get("key") or [""])[0] != BOSS_KEY: return self._json({"auth": 0}) try: data = get(action="getActiveActivations").get("data", []) ords = [{"id": o["activationId"], "phone": o["phoneNumber"], "service": o["serviceCode"], "status": int(o["activationStatus"]), "code": (o.get("smsCode") or "").strip()} for o in data] return self._json({"orders": ords, "balance": get(action="getBalance").split(":")[1]}) except Exception as e: return self._json({"error": str(e)}) if u.path == "/boss/api/store": if (q.get("key") or [""])[0] != BOSS_KEY: return self._json({"auth": 0}) st = store_load() ords = [] for k, v in sorted(st.items(), key=lambda x: str(x[0]), reverse=True): o = {"id": k, **v} o["flag"] = FLAGS.get(str(v.get("country")), "") if v.get("yuan") is None: try: p = get(action="getPrices") o["yuan"] = yuan(p[str(v["country"])][v["service"]]["cost"]) except Exception: o["yuan"] = "?" ords.append(o) return self._json({"orders": ords}) if u.path == "/boss/api/shot": if (q.get("key") or [""])[0] != BOSS_KEY: return self._json({"auth": 0}) o = store_load().get((q.get("id") or [""])[0], {}) sp = o.get("shot") if not sp or not os.path.exists(os.path.join(BASE, sp)): return self._json({"error": "该订单没有截图"}) with open(os.path.join(BASE, sp), "rb") as f: b = f.read() self.send_response(200) self.send_header("Content-Type", "image/png") self.send_header("Content-Length", str(len(b))) self.end_headers() self.wfile.write(b) return if u.path == "/boss/api/confirm": b = self._body() if b.get("key") != BOSS_KEY: return self._json({"auth": 0}) st = store_load() o = st.get(b.get("id", "")) if not o: return self._json({"error": "订单不存在"}) if o.get("spec") == "manual": # 成品号：手动发货，不自动买号 o["state"] = "active" o["manual"] = 1 store_save(st) return self._json({"ok": 1, "manual": 1}) try: r = get(action="getNumber", service=o["service"], country=o["country"]) if r.startswith("ACCESS_NUMBER"): _, myid, phone = r.split(":") o.update({"state": "active", "myid": myid, "phone": phone, "yuan": o.get("yuan") or yuan(get(action="getPrices")[str(o["country"])][o["service"]]["cost"]), "code": ""}) store_save(st) return self._json({"ok": 1, "myid": myid, "phone": phone}) return self._json({"error": {"NO_NUMBERS": "没库存了", "NO_BALANCE": "余额不足"}.get(r, r)}) except Exception as e: return self._json({"error": str(e)}) if u.path == "/boss/api/restore": b = self._body() if b.get("key") != BOSS_KEY: return self._json({"auth": 0}) st = store_load() o = st.get(b.get("id", "")) if o and o["state"] == "expired": o["state"] = "paid" o["ts"] = time.time() store_save(st) return self._json({"ok": 1}) if u.path == "/boss/api/reject": b = self._body() if b.get("key") != BOSS_KEY: return self._json({"auth": 0}) st = store_load() o = st.get(b.get("id", "")) if o: o["state"] = "rejected" store_save(st) return self._json({"ok": 1}) if u.path == "/boss/api/buy": b = self._body() if b.get("key") != BOSS_KEY: return self._json({"auth": 0}) try: r = get(action="getNumber", service=b["service"], country=b["country"]) if r.startswith("ACCESS_NUMBER"): _, aid, phone = r.split(":") return self._json({"ok": 1, "id": aid, "phone": phone}) return self._json({"error": {"NO_NUMBERS": "该国家没库存了", "NO_BALANCE": "余额不足，请充值"}.get(r, r)}) except Exception as e: return self._json({"error": str(e)}) if u.path == "/boss/api/cancel": b = self._body() if b.get("key") != BOSS_KEY: return self._json({"auth": 0}) try: r = get(action="setStatus", id=b["id"], status=8) return self._json({"ok": r == "ACCESS_STATUS_CHANGE"}) except Exception as e: return self._json({"error": str(e)}) if u.path == "/boss/api/qr": b = self._body() if b.get("key") != BOSS_KEY: return self._json({"auth": 0}) data = b.get("img", "") if "," not in data: return self._json({"error": "图片数据无效"}) with open(QR_FILE, "wb") as f: f.write(base64.b64decode(data.split(",", 1)[1])) return self._json({"ok": 1}) self._send(404, "Not Found") def do_POST(self): return self.do_GET() ALIAS = { "tg": "电报 tg telegram", "wa": "WhatsApp whats app", "ig": "Ins ins Instagram", "fb": "脸书 facebook", "go": "谷歌 google gmail", "lf": "抖音 TikTok", "tw": "推特 twitter x", "ot": "其他 other", "mb": "陌陌 momo", "mm": "微信 weixin wechat", "qq": "QQ", } print("✅ 小商店 v5.4 已启动") print(f" 本机访问 http://localhost:{PORT}") print(f" 客户页面 http://localhost:{PORT}/ （查码 | 下单 | 我的订单）") print(f" 老板后台 http://localhost:{PORT}/boss 密码: {BOSS_KEY}") print("停止服务：按 CTRL + C") import threading threading.Thread(target=preload, daemon=True).start() srv = ThreadingHTTPServer(("0.0.0.0", PORT), H) srv.serve_forever()