(self, task, tmp_dir)
| 1116 | task['progress'] = 100 |
| 1117 | browser.close() |
| 1118 | def _scrape_tiktok(self, task, tmp_dir): |
| 1119 | global core |
| 1120 | opts = task['options'] |
| 1121 | tutto = opts.get('tutto', False) |
| 1122 | target = (task.get('target') or "").strip() |
| 1123 | scrape_headers = self._scrape_headers("tiktok") |
| 1124 | req_session = requests.Session() |
| 1125 | req_session.headers.update(scrape_headers) |
| 1126 | tiktok_pagination = [] |
| 1127 | if target.startswith("https://") or target.startswith("http://"): |
| 1128 | if "tiktok.com" in target.lower(): |
| 1129 | target = target.split("tiktok.com", 1)[-1] |
| 1130 | target = target.split("?")[0] |
| 1131 | target = target.lstrip("/") |
| 1132 | if "/" in target: |
| 1133 | target = target.split("/")[0] |
| 1134 | if target.endswith("/"): |
| 1135 | target = target[:-1] |
| 1136 | if target.startswith("/"): |
| 1137 | target = target[1:] |
| 1138 | target = target.lstrip("@").strip() |
| 1139 | if not target: |
| 1140 | task['error'] = "Target TikTok vuoto" |
| 1141 | raise Exception("Target TikTok vuoto") |
| 1142 | target = "@" + target |
| 1143 | |
| 1144 | with sync_playwright() as p: |
| 1145 | browser = p.chromium.launch(headless=True, args=["--disable-blink-features=AutomationControlled", "--mute-audio"]) |
| 1146 | context = browser.new_context( |
| 1147 | user_agent=scrape_headers["User-Agent"], |
| 1148 | viewport={"width": 1920, "height": 1080} |
| 1149 | ) |
| 1150 | |
| 1151 | tiktok_sid = core.creds.get('tiktok_sid') |
| 1152 | if tiktok_sid: |
| 1153 | context.add_cookies([{"name": "sessionid", "value": tiktok_sid, "domain": ".tiktok.com", "path": "/"}]) |
| 1154 | |
| 1155 | page = context.new_page() |
| 1156 | |
| 1157 | # --- API Interception for raw video data --- |
| 1158 | api_videos = [] |
| 1159 | def _pick_video_id(item): |
| 1160 | return str(item.get("id") or item.get("aweme_id") or item.get("awemeId") or "").strip() |
| 1161 | |
| 1162 | def _extract_video_fields(item): |
| 1163 | video_data = item.get("video", {}) |
| 1164 | if not isinstance(video_data, dict): |
| 1165 | video_data = {} |
| 1166 | play_addr = ( |
| 1167 | video_data.get("playAddr") |
| 1168 | or video_data.get("downloadAddr") |
| 1169 | or video_data.get("play_addr") |
| 1170 | or video_data.get("download_addr") |
| 1171 | ) |
| 1172 | if isinstance(play_addr, dict): |
| 1173 | for key in ("urlList", "url_list"): |
| 1174 | url_list = play_addr.get(key) if isinstance(play_addr.get(key), list) else [] |
| 1175 | if url_list: |
no test coverage detected