socmind/bridges/tiktok_bridge.py
2026-09-11 23:33:20 -04:00

51 lines
1.8 KiB
Python

import asyncio
import json
import sys
async def run(req):
try:
from TikTokApi import TikTokApi
except ImportError:
return {"error": "TikTokApi not installed (pip install -r requirements.txt)"}
username = (req.get("username") or "").lstrip("@")
if not username:
return {"error": "username required"}
action = req.get("action", "user")
try:
async with TikTokApi() as api:
await api.create_sessions(num_sessions=1, sleep_after=3)
user = api.user(username)
info = await user.info()
stats = (info.get("stats") or {}) if isinstance(info, dict) else {}
base = {
"found": True, "platform": "tiktok", "username": username,
"followers": stats.get("followerCount"),
"following": stats.get("followingCount"),
"likes": stats.get("heartCount"),
"videos": stats.get("videoCount"),
"profileUrl": f"https://tiktok.com/@{username}",
}
if action == "videos":
count = int(req.get("count", 10) or 10)
vids = []
async for v in user.videos(count=count):
d = v.as_dict if hasattr(v, "as_dict") else {}
desc = (d.get("desc") or "")[:500]
vids.append({"id": str(d.get("id")), "desc": desc})
base["videos_list"] = vids
return base
except Exception as e:
return {"found": False, "username": username, "error": str(e)[:500]}
def main():
try:
req = json.loads(sys.stdin.read() or "{}")
except Exception as e:
print(json.dumps({"error": f"bad input: {e}"}))
return
print(json.dumps(asyncio.run(run(req))))
if __name__ == "__main__":
main()