# -*- coding: utf-8 -*- """ test.py 원격 Ubuntu 서버의 yellow.php + DB API를 직접 테스트하는 진단 도구입니다. 로컬 Windows PHP는 사용하지 않습니다. v43 기준: ig_media = 전역 게시물 마스터 ig_media_user_download = collector_id별 jpg/mp4 다운로드 상태 실행 예: py -3 test.py py -3 test.py -u yjm_pc_01 --ig test_account py -3 test.py -v """ import argparse import json import os import socket import urllib.parse import urllib.request from datetime import datetime BASE_URL = "https://n.yjm.kr/i/yellow.php" TOKEN = "ABC1234XYZ" def default_user(): return (os.environ.get("COMPUTERNAME") or socket.gethostname() or "local_pc").strip() or "local_pc" def mask_secret(value): s = str(value or "") if len(s) <= 8: return "****" if s else "" return s[:4] + "..." + s[-3:] def call_api(base_url, token, mode, data=None, verbose=False): payload = {"token": token, "mode": mode} if data: payload.update(data) log_payload = dict(payload) log_payload["token"] = mask_secret(log_payload.get("token")) body = urllib.parse.urlencode(payload).encode("utf-8") req = urllib.request.Request( base_url, data=body, headers={"Content-Type": "application/x-www-form-urlencoded; charset=utf-8"}, method="POST", ) print("\n==============================") print("MODE:", mode) if verbose: print("URL:", base_url) print("PAYLOAD:", json.dumps(log_payload, ensure_ascii=False, indent=2)) try: with urllib.request.urlopen(req, timeout=10) as res: raw = res.read().decode("utf-8", errors="replace") print("HTTP:", getattr(res, "status", "")) print("RAW:", raw) try: parsed = json.loads(raw) print("JSON:") print(json.dumps(parsed, ensure_ascii=False, indent=2)) return parsed except Exception as e: print("JSON parse error:", e) return {"ok": 0, "error": "bad_json", "raw": raw} except Exception as e: print("HTTP/API error:", e) return {"ok": 0, "error": str(e)} def main(argv=None): parser = argparse.ArgumentParser(description="yellow.php remote API diagnostic test") parser.add_argument("--url", default=BASE_URL, help="yellow.php URL") parser.add_argument("--token", default=TOKEN, help="yellow.php API token") parser.add_argument("-u", "--user", dest="collector_id", default="", help="수집 사용자/작업자 ID. 생략하면 컴퓨터 이름") parser.add_argument("--ig", dest="ig_account_id", default="test_account", help="테스트용 인스타 계정 ID") parser.add_argument("-v", "--verbose", action="store_true", help="요청 payload를 자세히 출력") parser.add_argument("--show-schema", action="store_true", help="서버의 yellow_schema_reset.sql 내용을 출력") args = parser.parse_args(argv) collector_id = (args.collector_id or default_user()).strip() ig_account_id = args.ig.strip() or "test_account" shortcode = "TEST_" + datetime.now().strftime("%Y%m%d_%H%M%S") save_folder = rf"C:\Users\{collector_id}\Downloads\Down\{collector_id}\{ig_account_id}" print("yellow.php remote API test") print("collector_id:", collector_id) print("ig_account_id:", ig_account_id) print("token:", mask_secret(args.token)) common = { "collector_id": collector_id, "user_id": collector_id, "ig_account_id": ig_account_id, "account_name": ig_account_id, "folder_name": ig_account_id, "save_folder": save_folder, } call_api(args.url, args.token, "health", verbose=args.verbose) call_api(args.url, args.token, "token", verbose=args.verbose) if args.show_schema: call_api(args.url, args.token, "schema_sql", verbose=args.verbose) # 1. 존재 여부 확인: 새 shortcode라면 NEW call_api(args.url, args.token, "media_exists", {**common, "shortcode": shortcode}, verbose=args.verbose) # 2. jpg/thumbnail 다운로드 완료를 먼저 기록: user row는 partial이어야 정상 call_api(args.url, args.token, "media_insert", { **common, "shortcode": shortcode, "media_url": f"https://www.instagram.com/reel/{shortcode}/", "upload_at": "2026-06-24 20:00:00", "asset_kind": "image", "asset_status": "done", "file_name": f"{shortcode}.jpg", "file_path": save_folder + rf"\{shortcode}.jpg", "md5": "d41d8cd98f00b204e9800998ecf8427e", "sha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", "file_size": 12345, "views": 100, "likes": 10, "comments": 2, "reposts": 1, }, verbose=args.verbose) # 3. jpg만 있는 상태 확인: exists=0, status=PARTIAL 이어야 재시도/MP4 시도가 가능 call_api(args.url, args.token, "media_exists", {**common, "shortcode": shortcode}, verbose=args.verbose) # 4. mp4 다운로드 완료 기록: jpg+mp4가 모두 done이면 complete/PSSS가 되어야 정상 call_api(args.url, args.token, "media_insert", { **common, "shortcode": shortcode, "media_url": f"https://www.instagram.com/reel/{shortcode}/", "upload_at": "2026-06-24 20:00:00", "asset_kind": "video", "asset_status": "done", "file_name": f"{shortcode}.mp4", "file_path": save_folder + rf"\{shortcode}.mp4", "md5": "0cc175b9c0f1b6a831c399e269772661", "sha256": "ca978112ca1bbdcafac231b39a23dc4da786eff8147c4e72b9807785afee48bb", "file_size": 987654, "views": 200, "likes": 20, "comments": 5, "reposts": 2, }, verbose=args.verbose) # 5. 완성 상태 확인: exists=1, status=PSSS, download_status=complete 이어야 정상 call_api(args.url, args.token, "media_exists", {**common, "shortcode": shortcode}, verbose=args.verbose) # 6. master 최신값 갱신 테스트 call_api(args.url, args.token, "media_update", { **common, "shortcode": shortcode, "views": 300, "likes": 30, "comments": 7, "reposts": 3, }, verbose=args.verbose) # 7. 사용자 + 인스타 ID별 최근 저장 목록 call_api(args.url, args.token, "media_list", {**common, "limit": 10}, verbose=args.verbose) # 8. complete만 필터 call_api(args.url, args.token, "media_list", {**common, "download_status": "complete", "limit": 10}, verbose=args.verbose) # 9. stats / history call_api(args.url, args.token, "media_stats", {"collector_id": collector_id, "user_id": collector_id}, verbose=args.verbose) call_api(args.url, args.token, "media_history", {**common, "shortcode": shortcode}, verbose=args.verbose) if __name__ == "__main__": main()