tiktok根据视频url获取数据

├── main.py              # Script entry file
├── requirements.txt     # Python dependency management
├── input_schema.json    # UI input form configuration
├── README.md            # Project documentation
├── sdk.py               # SDK core functions
├── sdk_pb2.py           # Enhanced data processing module
├── sdk_pb2_grpc.py      # Network communication module

# Retrieve all input parameters as a dictionary
parameters = SDK.Parameter.get_input_json_dict()

# Example: Assuming website URL and keywords are provided
# Returns: {"website": "example.com", "keyword": "news"}

SDK.Log.debug("Connecting to target website...")           # Debug info
SDK.Log.info("Successfully retrieved 10 news items")       # Normal process log
SDK.Log.warn("Network is slow, may affect collection speed")  # Warning
SDK.Log.error("Cannot access target website, check network")  # Error

headers = [
    {
        "label": "News Title",     # Column name (user-visible)
        "key": "title",           # Column key (used in code)
        "format": "text",         # Data type
    },
    {
        "label": "Publish Time",
        "key": "publish_time",
        "format": "text",
    },
    {
        "label": "News Category",
        "key": "category",
        "format": "text",
    },
]

res = CoreSDK.Result.set_table_header(headers)

news_data = [
    {"title": "AI Breakthrough", "publish_time": "2023-10-01", "category": "Tech"},
    {"title": "Stock Market Trends", "publish_time": "2023-10-01", "category": "Finance"},
]

for i, news in enumerate(news_data):
    obj = {
        "title": news.get('title'),
        "publish_time": news.get('publish_time'),
        "category": news.get('category'),
    }
    res = CoreSDK.Result.push_data(obj)
    SDK.Log.info(f"Pushed data item {i+1}: {news.get('title')}")

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import asyncio
import os

from sdk import CoreSDK

async def run():
    try:
        # 1. Get startup parameters
        config = SDK.Parameter.get_input_json_dict()
        website = config.get("website", "Default website")
        SDK.Log.info(f"Starting data collection from: {website}")

        # 2. Set result table headers
        headers = [
            {"label": "Title", "key": "title", "format": "text"},
            {"label": "Time", "key": "publish_time", "format": "text"},
            {"label": "Category", "key": "category", "format": "text"},
            {"label": "View Count", "key": "view_count", "format": "integer"},
        ]
        CoreSDK.Result.set_table_header(headers)

        # 3. Simulate data collection (replace with actual scraping code)
        collected_data = [
            {"title": "Sample News 1", "publish_time": "2023-10-01 10:00", "category": "Tech", "view_count": 1000},
            {"title": "Sample News 2", "publish_time": "2023-10-01 11:00", "category": "Finance", "view_count": 500},
        ]

        # 4. Push data
        for data in collected_data:
            obj = {
                "title": data.get("title"),
                "publish_time": data.get("publish_time"),
                "category": data.get("category"),
                "view_count": data.get("view_count", 0),
            }
            res = CoreSDK.Result.push_data(obj)

        # 5. Complete
        SDK.Log.info("Data collection task completed!")

    except Exception as e:
        SDK.Log.error(f"Script execution error: {e}")
        error_result = {
            "error": str(e),
            "error_code": "500",
            "status": "failed"
        }
        CoreSDK.Result.push_data(error_result)
        raise

if __name__ == "__main__":
    asyncio.run(run())

attrs==25.4.0
beautifulsoup4==4.14.2
certifi==2025.10.5
cffi==2.0.0
charset-normalizer==3.4.4
click==8.3.0
colorama==0.4.6
cssselect==1.3.0
DataRecorder==3.6.2
DownloadKit==2.0.7
DrissionPage==4.1.1.2
et_xmlfile
filelock