| 970 | # ── VPS deployment ─────────────────────────────────────────────────── |
| 971 | |
| 972 | def deploy_reddit_post( |
| 973 | reddit_data: Dict, |
| 974 | vps_host: str, |
| 975 | vps_user: str, |
| 976 | pkey, |
| 977 | ) -> bool: |
| 978 | import paramiko |
| 979 | |
| 980 | title = reddit_data.get("title", "") |
| 981 | image_urls = get_post_image_urls(reddit_data) |
| 982 | image_url = image_urls[0] if image_urls else "" |
| 983 | |
| 984 | if not all([title, image_url, vps_host, vps_user, pkey]): |
| 985 | print(" VPS: Missing required arguments") |
| 986 | return False |
| 987 | |
| 988 | try: |
| 989 | ssh = paramiko.SSHClient() |
| 990 | ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) |
| 991 | |
| 992 | print(f" VPS: Connecting to {vps_user}@{vps_host}...") |
| 993 | |
| 994 | ssh.connect( |
| 995 | hostname=vps_host, |
| 996 | username=vps_user, |
| 997 | pkey=pkey, |
| 998 | timeout=30, |
| 999 | allow_agent=False, |
| 1000 | look_for_keys=False, |
| 1001 | ) |
| 1002 | |
| 1003 | base_cmd = "cd /root/reddit_post_automation" |
| 1004 | |
| 1005 | # Sanitize for unquoted heredoc (shell expansion active) |
| 1006 | # Also strip newlines to prevent breaking TypeScript syntax |
| 1007 | def _sanitize(s: str) -> str: |
| 1008 | return (s |
| 1009 | .replace('\n', ' ').replace('\r', '') |
| 1010 | .replace('LINKEOF', '') |
| 1011 | .replace('\\', '\\\\') |
| 1012 | .replace('"', '\\"') |
| 1013 | .replace('$', '\\$') |
| 1014 | .replace('`', '\\`')) |
| 1015 | |
| 1016 | sanitized_url = _sanitize(image_url) |
| 1017 | sanitized_title = _sanitize(title) |
| 1018 | |
| 1019 | update_link_cmd = f"""{base_cmd} && cat > src/link.ts << LINKEOF |
| 1020 | const LINK = "{sanitized_url}"; |
| 1021 | const TITLE = "{sanitized_title}"; |
| 1022 | export {{LINK, TITLE}}; |
| 1023 | LINKEOF |
| 1024 | """ |
| 1025 | |
| 1026 | print(" VPS: Updating link.ts...") |
| 1027 | _stdin, _stdout, stderr = ssh.exec_command(update_link_cmd) |
| 1028 | err = stderr.read().decode().strip() |
| 1029 | if err: |