diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..75f7f4a --- /dev/null +++ b/.env.example @@ -0,0 +1,13 @@ +# .env.example + +# ACCESS_TOKEN is used for authenticating API requests to Meta's Threads app and Threads.net, +# Required: No default. +#ACCESS_TOKEN="" + +# BASE_URL specifies the API endpoint for accessing Meta's Threads app. +# Optional: Default= "https://graph.threads.net/v1.0" +#BASE_URL="https://graph.threads.net/v1.0" + +# DRAFTS_FILE specifies the file name or path for saving draft data. +# Optional: Default = "drafts.json" +#DRAFTS_FILE="drafts.json" diff --git a/.github/workflows/python-app.yml b/.github/workflows/python-app.yml index c19ce10..e9eb30d 100644 --- a/.github/workflows/python-app.yml +++ b/.github/workflows/python-app.yml @@ -39,5 +39,7 @@ jobs: # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics - name: Test with pytest + env: + ACCESS_TOKEN: ${{ secrets.ACCESS_TOKEN }} run: | pytest diff --git a/main.py b/main.py index ae517db..bdc9249 100644 --- a/main.py +++ b/main.py @@ -1,323 +1,5 @@ -import typer -import os -import threading -import json -from rich.console import Console -from rich.table import Table -from datetime import datetime, timedelta, timezone -from dotenv import load_dotenv -from api import get_user_id, get_user_profile, get_user_posts, get_post_insights, fetch_all_posts, create_post, get_post_replies, get_post_replies_count -from utils import convert_to_locale - -app = typer.Typer() -console = Console() -load_dotenv() - -ACCESS_TOKEN = os.getenv("ACCESS_TOKEN") -HEADERS = { - 'Authorization': f'Bearer {ACCESS_TOKEN}' -} -DRAFTS_FILE = 'drafts.json' -SERVER_PROCESS_TIME = 10 - -@app.command() -def get_profile(): - """ - Retrieve and display user profile information, including the last post made by the user. - """ - user_id = get_user_id(HEADERS) - profile = get_user_profile(user_id, HEADERS) - last_post = get_user_posts(user_id, HEADERS, limit=1)[0] - - profile_table = Table(title=f'{profile["username"]}\'s Profile') - profile_table.add_column("Field", style="cyan", no_wrap=True) - profile_table.add_column("Value", style="magenta") - - profile_table.add_row("ID", profile.get("id", "N/A")) - profile_table.add_row("Username", profile.get("username", "N/A")) - profile_table.add_row("Profile Picture URL", profile.get("threads_profile_picture_url", "N/A")) - profile_table.add_row("Biography", profile.get("threads_biography", "N/A")) - if last_post: - profile_table.add_row("Last Post ID", last_post.get("id", "N/A")) - profile_table.add_row("Post Type", last_post.get("media_type", "N/A")) - profile_table.add_row("Post Text", last_post.get("text", "N/A")) - profile_table.add_row("Post Permalink", last_post.get("permalink", "N/A")) - profile_table.add_row("Post Timestamp", convert_to_locale(last_post.get("timestamp", "N/A"))) - else: - profile_table.add_row("Message", "No posts found") - - console.print(profile_table) - -@app.command() -def get_recent_posts(limit: int = 5): - """ - Retrieve the most recent posts. - """ - user_id = get_user_id(HEADERS) - posts = get_user_posts(user_id, HEADERS, limit=limit) - - table = Table(title="Recent Posts") - table.add_column("ID", style="cyan", no_wrap=True) - table.add_column("Username", style="cyan", no_wrap=True) - table.add_column("Timestamp", style="magenta") - table.add_column("Type", style="green") - table.add_column("Text", style="yellow") - table.add_column("Permalink", style="blue") - table.add_column("Replies", style="red") - - for post in posts: - if post.get('media_type') == 'REPOST_FACADE': - continue - timestamp = convert_to_locale(post.get('timestamp', 'N/A')) - replies_count = get_post_replies_count(post['id'], HEADERS) - table.add_row( - post.get('id', 'N/A'), - post.get('username', 'N/A'), - timestamp, - post.get('media_type', 'N/A'), - post.get('text', 'N/A'), - post.get('permalink', 'N/A'), - str(replies_count) - ) - - console.print(table) - -@app.command() -def get_top_liked_posts(limit: int = 5, time_range: str = None): - """ - Retrieve the top liked posts of all time or within a specific time range. - """ - user_id = get_user_id(HEADERS) - all_posts = fetch_all_posts(user_id, HEADERS) - - if time_range: - now = datetime.now(timezone.utc) - if time_range.endswith('w'): - weeks = int(time_range[:-1]) - start_time = now - timedelta(weeks=weeks) - elif time_range.endswith('d'): - days = int(time_range[:-1]) - start_time = now - timedelta(days=days) - elif time_range.endswith('h'): - hours = int(time_range[:-1]) - start_time = now - timedelta(hours=hours) - elif time_range.endswith('m'): - months = int(time_range[:-1]) - start_time = now - timedelta(days=30 * months) - else: - typer.echo("Invalid time range format. Use '2w' for 2 weeks, '7d' for 7 days, '24h' for 24 hours, or '7m' for 7 months.") - return - - all_posts = [post for post in all_posts if datetime.strptime(post['timestamp'], '%Y-%m-%dT%H:%M:%S%z') >= start_time] - - posts_with_likes = [] - for post in all_posts: - if post.get('media_type') == 'REPOST_FACADE': - continue - insights = get_post_insights(post['id'], HEADERS) - if 'likes' in insights: - posts_with_likes.append((post, insights['likes'])) - - posts_with_likes.sort(key=lambda x: x[1], reverse=True) - top_liked_posts = posts_with_likes[:limit] - - table = Table(title="Top Liked Posts") - table.add_column("Username", style="cyan", no_wrap=True) - table.add_column("Timestamp", style="magenta") - table.add_column("Type", style="green") - table.add_column("Text", style="yellow") - table.add_column("Permalink", style="blue") - table.add_column("Likes", style="red") - table.add_column("Replies", style="green") - table.add_column("Reposts", style="blue") - table.add_column("Quotes", style="yellow") - table.add_column("Views", style="cyan") - - for post, likes in top_liked_posts: - timestamp = convert_to_locale(post.get('timestamp', 'N/A')) - insights = get_post_insights(post['id'], HEADERS) - table.add_row( - post.get('username', 'N/A'), - timestamp, - post.get('media_type', 'N/A'), - post.get('text', 'N/A'), - post.get('permalink', 'N/A'), - str(insights.get('likes', 'N/A')), - str(insights.get('replies', 'N/A')), - str(insights.get('reposts', 'N/A')), - str(insights.get('quotes', 'N/A')), - str(insights.get('views', 'N/A')) - ) - - console.print(table) - -@app.command() -def create_text_post(text: str): - """ - Create a post with text. - """ - user_id = get_user_id(HEADERS) - payload = { - "media_type": "TEXT", - "text": text - } - post = create_post(user_id, HEADERS, payload) - typer.echo(f"Post created with ID: {post['id']}") - -@app.command() -def create_image_post(text: str, image_url: str): - """ - Create a post with an image. - """ - user_id = get_user_id(HEADERS) - payload = { - "media_type": "IMAGE", - "image_url": image_url, - "text": text - } - post = create_post(user_id, HEADERS, payload) - typer.echo(f"Post created with ID: {post['id']}") - -@app.command() -def get_latest_replies(media_id: str, limit: int = 5): - """ - Retrieve the latest replies for a specific media post. - """ - replies = get_post_replies(media_id, HEADERS, limit=limit) - - table = Table(title="Latest Replies") - table.add_column("Username", style="cyan", no_wrap=True) - table.add_column("Media ID", style="cyan", no_wrap=True) - table.add_column("Timestamp", style="magenta") - table.add_column("Text", style="yellow") - table.add_column("Permalink", style="blue") - - for reply in replies: - timestamp = convert_to_locale(reply.get('timestamp', 'N/A')) - table.add_row( - reply.get('username', 'N/A'), - reply.get('id', 'N/A'), - timestamp, - reply.get('text', 'N/A'), - reply.get('permalink', 'N/A') - ) - - console.print(table) - -@app.command() -def send_reply(media_id: str, text: str): - """ - Send a reply to a specific media post. - """ - user_id = get_user_id(HEADERS) - payload = { - "media_type": "TEXT", - "text": text, - "reply_to_id": media_id - } - reply = create_post(user_id, HEADERS, payload) - typer.echo(f"Reply created with ID: {reply['id']}") - -def job_create_text_post(text: str): - """ - Job function to create a post with text. - """ - create_text_post(text) - -@app.command() -def schedule_post(text: str, post_time: str): - """ - Schedule a post with text at a specific time. - """ - post_time_dt = datetime.strptime(post_time, '%Y-%m-%d %H:%M:%S') - current_time = datetime.now() - delay = (post_time_dt - current_time).total_seconds() - - if delay <= 0: - typer.echo("Scheduled time must be in the future.") - return - - timer = threading.Timer(delay, job_create_text_post, [text]) - timer.start() - typer.echo(f"Post scheduled for {post_time} with text: '{text}'") - -@app.command() -def create_draft(text: str, drafts_file: str = DRAFTS_FILE): - ''' - Create a draft with the given text and save it to the drafts file. - ''' - if os.path.exists(drafts_file): - with open(drafts_file, 'r') as file: - drafts = json.load(file) - else: - drafts = [] - - next_id = max([draft['id'] for draft in drafts], default=0) + 1 - - draft = { - "id": next_id, - "text": text, - "timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S') - } - drafts.append(draft) - - with open(drafts_file, 'w') as file: - json.dump(drafts, file, indent=4) - - typer.echo(f"Draft created with ID: {next_id}") - -@app.command() -def get_drafts(drafts_file: str = DRAFTS_FILE): - ''' - Get all drafts from the drafts file. - ''' - if not os.path.exists(drafts_file): - typer.echo("No drafts found.") - return - - with open(drafts_file, 'r') as file: - drafts = json.load(file) - - table = Table(title="Drafts") - table.add_column("ID", style="cyan", no_wrap=True) - table.add_column("Text", style="yellow") - table.add_column("Timestamp", style="magenta") - - for draft in drafts: - table.add_row( - str(draft['id']), - draft['text'], - draft['timestamp'] - ) - - console.print(table) - -@app.command() -def send_draft(draft_id: int, drafts_file: str = DRAFTS_FILE): - ''' - Send a draft with the given ID and remove it from the drafts file. - ''' - if not os.path.exists(drafts_file): - typer.echo("No drafts found.") - raise typer.Exit(1) - - with open(drafts_file, 'r') as file: - drafts = json.load(file) - - draft = next((draft for draft in drafts if draft['id'] == draft_id), None) - - if draft is None: - typer.echo(f"Draft with ID {draft_id} not found.") - raise typer.Exit(1) - - create_text_post(draft['text']) - - drafts = [draft for draft in drafts if draft['id'] != draft_id] - - with open(drafts_file, 'w') as file: - json.dump(drafts, file, indent=4) - - typer.echo(f"Draft with ID {draft_id} sent and removed from drafts.") +import src.env +from src.app import app def main(): app() diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/__init__.py @@ -0,0 +1 @@ + diff --git a/api.py b/src/api.py similarity index 96% rename from api.py rename to src/api.py index 0e5df89..673b93a 100644 --- a/api.py +++ b/src/api.py @@ -1,7 +1,7 @@ import requests -from utils import convert_to_locale -BASE_URL = "https://graph.threads.net/v1.0" +from .env import BASE_URL +from .utils import convert_to_locale def get_user_id(headers): response = requests.get(f"{BASE_URL}/me?fields=id", headers=headers) @@ -63,4 +63,4 @@ def get_post_replies_count(post_id, headers): response = requests.get(f"{BASE_URL}/{post_id}/replies", headers=headers) response.raise_for_status() replies = response.json().get('data', []) - return len(replies) \ No newline at end of file + return len(replies) diff --git a/src/app.py b/src/app.py new file mode 100644 index 0000000..de693eb --- /dev/null +++ b/src/app.py @@ -0,0 +1,339 @@ +import os +import sys +import threading +from datetime import datetime, timedelta, timezone +import json + +import typer +from rich.console import Console +from rich.table import Table + +from .env import ACCESS_TOKEN, DRAFTS_FILE +from .api import ( + get_user_id, + get_user_profile, + get_user_posts, + get_post_insights, + fetch_all_posts, + create_post, + get_post_replies, + get_post_replies_count, +) +from .utils import convert_to_locale +from .draft_utils import ensure_drafts_file + +ACCESS_TOKEN = os.getenv("ACCESS_TOKEN") + +if not ACCESS_TOKEN: + print("Error: The required ACCESS_TOKEN is not set. Please set it in your environment or in your .env file.") + sys.exit(1) + +# Ensure DRAFTS_FILE path is resolved and the file exists, following XDG Base Directory Specification if necessary. +DRAFTS_FILE = ensure_drafts_file(DRAFTS_FILE) + +HEADERS = { + 'Authorization': f'Bearer {ACCESS_TOKEN}' +} +SERVER_PROCESS_TIME = 10 + +app = typer.Typer() +console = Console() + +@app.command() +def get_profile(): + """ + Retrieve and display user profile information, including the last post made by the user. + """ + user_id = get_user_id(HEADERS) + profile = get_user_profile(user_id, HEADERS) + last_post = get_user_posts(user_id, HEADERS, limit=1)[0] + + profile_table = Table(title=f'{profile["username"]}\'s Profile') + profile_table.add_column("Field", style="cyan", no_wrap=True) + profile_table.add_column("Value", style="magenta") + + profile_table.add_row("ID", profile.get("id", "N/A")) + profile_table.add_row("Username", profile.get("username", "N/A")) + profile_table.add_row("Profile Picture URL", profile.get("threads_profile_picture_url", "N/A")) + profile_table.add_row("Biography", profile.get("threads_biography", "N/A")) + if last_post: + profile_table.add_row("Last Post ID", last_post.get("id", "N/A")) + profile_table.add_row("Post Type", last_post.get("media_type", "N/A")) + profile_table.add_row("Post Text", last_post.get("text", "N/A")) + profile_table.add_row("Post Permalink", last_post.get("permalink", "N/A")) + profile_table.add_row("Post Timestamp", convert_to_locale(last_post.get("timestamp", "N/A"))) + else: + profile_table.add_row("Message", "No posts found") + + console.print(profile_table) + +@app.command() +def get_recent_posts(limit: int = 5): + """ + Retrieve the most recent posts. + """ + user_id = get_user_id(HEADERS) + posts = get_user_posts(user_id, HEADERS, limit=limit) + + table = Table(title="Recent Posts") + table.add_column("ID", style="cyan", no_wrap=True) + table.add_column("Username", style="cyan", no_wrap=True) + table.add_column("Timestamp", style="magenta") + table.add_column("Type", style="green") + table.add_column("Text", style="yellow") + table.add_column("Permalink", style="blue") + table.add_column("Replies", style="red") + + for post in posts: + if post.get('media_type') == 'REPOST_FACADE': + continue + timestamp = convert_to_locale(post.get('timestamp', 'N/A')) + replies_count = get_post_replies_count(post['id'], HEADERS) + table.add_row( + post.get('id', 'N/A'), + post.get('username', 'N/A'), + timestamp, + post.get('media_type', 'N/A'), + post.get('text', 'N/A'), + post.get('permalink', 'N/A'), + str(replies_count) + ) + + console.print(table) + +@app.command() +def get_top_liked_posts(limit: int = 5, time_range: str = None): + """ + Retrieve the top liked posts of all time or within a specific time range. + """ + user_id = get_user_id(HEADERS) + all_posts = fetch_all_posts(user_id, HEADERS) + + if time_range: + now = datetime.now(timezone.utc) + if time_range.endswith('w'): + weeks = int(time_range[:-1]) + start_time = now - timedelta(weeks=weeks) + elif time_range.endswith('d'): + days = int(time_range[:-1]) + start_time = now - timedelta(days=days) + elif time_range.endswith('h'): + hours = int(time_range[:-1]) + start_time = now - timedelta(hours=hours) + elif time_range.endswith('m'): + months = int(time_range[:-1]) + start_time = now - timedelta(days=30 * months) + else: + typer.echo("Invalid time range format. Use '2w' for 2 weeks, '7d' for 7 days, '24h' for 24 hours, or '7m' for 7 months.") + return + + all_posts = [post for post in all_posts if datetime.strptime(post['timestamp'], '%Y-%m-%dT%H:%M:%S%z') >= start_time] + + posts_with_likes = [] + for post in all_posts: + if post.get('media_type') == 'REPOST_FACADE': + continue + insights = get_post_insights(post['id'], HEADERS) + if 'likes' in insights: + posts_with_likes.append((post, insights['likes'])) + + posts_with_likes.sort(key=lambda x: x[1], reverse=True) + top_liked_posts = posts_with_likes[:limit] + + table = Table(title="Top Liked Posts") + table.add_column("Username", style="cyan", no_wrap=True) + table.add_column("Timestamp", style="magenta") + table.add_column("Type", style="green") + table.add_column("Text", style="yellow") + table.add_column("Permalink", style="blue") + table.add_column("Likes", style="red") + table.add_column("Replies", style="green") + table.add_column("Reposts", style="blue") + table.add_column("Quotes", style="yellow") + table.add_column("Views", style="cyan") + + for post, likes in top_liked_posts: + timestamp = convert_to_locale(post.get('timestamp', 'N/A')) + insights = get_post_insights(post['id'], HEADERS) + table.add_row( + post.get('username', 'N/A'), + timestamp, + post.get('media_type', 'N/A'), + post.get('text', 'N/A'), + post.get('permalink', 'N/A'), + str(insights.get('likes', 'N/A')), + str(insights.get('replies', 'N/A')), + str(insights.get('reposts', 'N/A')), + str(insights.get('quotes', 'N/A')), + str(insights.get('views', 'N/A')) + ) + + console.print(table) + +@app.command() +def create_text_post(text: str): + """ + Create a post with text. + """ + user_id = get_user_id(HEADERS) + payload = { + "media_type": "TEXT", + "text": text + } + post = create_post(user_id, HEADERS, payload) + typer.echo(f"Post created with ID: {post['id']}") + +@app.command() +def create_image_post(text: str, image_url: str): + """ + Create a post with an image. + """ + user_id = get_user_id(HEADERS) + payload = { + "media_type": "IMAGE", + "image_url": image_url, + "text": text + } + post = create_post(user_id, HEADERS, payload) + typer.echo(f"Post created with ID: {post['id']}") + +@app.command() +def get_latest_replies(media_id: str, limit: int = 5): + """ + Retrieve the latest replies for a specific media post. + """ + replies = get_post_replies(media_id, HEADERS, limit=limit) + + table = Table(title="Latest Replies") + table.add_column("Username", style="cyan", no_wrap=True) + table.add_column("Media ID", style="cyan", no_wrap=True) + table.add_column("Timestamp", style="magenta") + table.add_column("Text", style="yellow") + table.add_column("Permalink", style="blue") + + for reply in replies: + timestamp = convert_to_locale(reply.get('timestamp', 'N/A')) + table.add_row( + reply.get('username', 'N/A'), + reply.get('id', 'N/A'), + timestamp, + reply.get('text', 'N/A'), + reply.get('permalink', 'N/A') + ) + + console.print(table) + +@app.command() +def send_reply(media_id: str, text: str): + """ + Send a reply to a specific media post. + """ + user_id = get_user_id(HEADERS) + payload = { + "media_type": "TEXT", + "text": text, + "reply_to_id": media_id + } + reply = create_post(user_id, HEADERS, payload) + typer.echo(f"Reply created with ID: {reply['id']}") + +def job_create_text_post(text: str): + """ + Job function to create a post with text. + """ + create_text_post(text) + +@app.command() +def schedule_post(text: str, post_time: str): + """ + Schedule a post with text at a specific time. + """ + post_time_dt = datetime.strptime(post_time, '%Y-%m-%d %H:%M:%S') + current_time = datetime.now() + delay = (post_time_dt - current_time).total_seconds() + + if delay <= 0: + typer.echo("Scheduled time must be in the future.") + return + + timer = threading.Timer(delay, job_create_text_post, [text]) + timer.start() + typer.echo(f"Post scheduled for {post_time} with text: '{text}'") + +@app.command() +def create_draft(text: str, drafts_file: str = DRAFTS_FILE): + ''' + Create a draft with the given text and save it to the drafts file. + ''' + if os.path.exists(drafts_file): + with open(drafts_file, 'r') as file: + drafts = json.load(file) + else: + drafts = [] + + next_id = max([draft['id'] for draft in drafts], default=0) + 1 + + draft = { + "id": next_id, + "text": text, + "timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S') + } + drafts.append(draft) + + with open(drafts_file, 'w') as file: + json.dump(drafts, file, indent=4) + + typer.echo(f"Draft created with ID: {next_id}") + +@app.command() +def get_drafts(drafts_file: str = DRAFTS_FILE): + ''' + Get all drafts from the drafts file. + ''' + if not os.path.exists(drafts_file): + typer.echo("No drafts found.") + return + + with open(drafts_file, 'r') as file: + drafts = json.load(file) + + table = Table(title="Drafts") + table.add_column("ID", style="cyan", no_wrap=True) + table.add_column("Text", style="yellow") + table.add_column("Timestamp", style="magenta") + + for draft in drafts: + table.add_row( + str(draft['id']), + draft['text'], + draft['timestamp'] + ) + + console.print(table) + +@app.command() +def send_draft(draft_id: int, drafts_file: str = DRAFTS_FILE): + ''' + Send a draft with the given ID and remove it from the drafts file. + ''' + if not os.path.exists(drafts_file): + typer.echo("No drafts found.") + raise typer.Exit(1) + + with open(drafts_file, 'r') as file: + drafts = json.load(file) + + draft = next((draft for draft in drafts if draft['id'] == draft_id), None) + + if draft is None: + typer.echo(f"Draft with ID {draft_id} not found.") + raise typer.Exit(1) + + create_text_post(draft['text']) + + drafts = [draft for draft in drafts if draft['id'] != draft_id] + + with open(drafts_file, 'w') as file: + json.dump(drafts, file, indent=4) + + typer.echo(f"Draft with ID {draft_id} sent and removed from drafts.") diff --git a/src/draft_utils.py b/src/draft_utils.py new file mode 100644 index 0000000..d3ef53c --- /dev/null +++ b/src/draft_utils.py @@ -0,0 +1,46 @@ +import os +import json + +def ensure_drafts_file(drafts_file: str) -> str: + """ + Ensure the drafts file exists following these rules: + + - If the provided drafts_file does not exist and it is a simple filename (i.e., it does not contain + any path separator), then use the XDG Base Directory Specification to determine a cache directory: + * It checks for the XDG_CACHE_HOME environment variable. + * If not set, defaults to HOME/.cache. + Then, a sub-folder named "threads-cli" is created within that cache directory, and + drafts_file is placed inside that sub-folder. + + - If drafts_file is a simple filename, but a file does not exist at that location, or if it is given as a full + path, the application uses that path exactly as given. + + - If the drafts file does not exist, create an empty JSON file (with an empty dict as content by default). + + Returns: + The final path to the drafts file. + + Raises: + EnvironmentError: If the required HOME environment variable is not set when needed. + """ + # Check if drafts_file doesn't exist and is a simple filename (without path separator). + if not os.path.exists(drafts_file) and os.path.sep not in drafts_file: + # Determine the cache directory using XDG_CACHE_HOME if available, + # otherwise default to HOME/.cache. + xdg_cache_home = os.getenv('XDG_CACHE_HOME') + if not xdg_cache_home: + home = os.getenv('HOME') + if not home: + raise EnvironmentError("HOME environment variable is not set.") + xdg_cache_home = os.path.join(home, '.cache') + # Define the final path: XDG_CACHE_HOME/threads-cli/drafts_file + drafts_dir = os.path.join(xdg_cache_home, "threads-cli") + os.makedirs(drafts_dir, exist_ok=True) + drafts_file = os.path.join(drafts_dir, drafts_file) + + # If the drafts file still does not exist, create an empty JSON file. + if not os.path.exists(drafts_file): + with open(drafts_file, "w") as f: + json.dump({}, f) + + return drafts_file diff --git a/src/env.py b/src/env.py new file mode 100644 index 0000000..9c56374 --- /dev/null +++ b/src/env.py @@ -0,0 +1,10 @@ +import os + +from dotenv import load_dotenv + +load_dotenv() + +ACCESS_TOKEN = os.getenv("ACCESS_TOKEN", "") +BASE_URL = os.getenv("BASE_URL", "https://graph.threads.net/v1.0") + +DRAFTS_FILE = os.getenv("DRAFTS_FILE", "drafts.json") diff --git a/utils.py b/src/utils.py similarity index 85% rename from utils.py rename to src/utils.py index 3117e82..7f1063b 100644 --- a/utils.py +++ b/src/utils.py @@ -7,4 +7,4 @@ def convert_to_locale(timestamp): if timestamp == 'N/A': return timestamp dt = datetime.strptime(timestamp, '%Y-%m-%dT%H:%M:%S%z') - return dt.strftime('%Y-%m-%d %H:%M:%S') \ No newline at end of file + return dt.strftime('%Y-%m-%d %H:%M:%S') diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..26ff74e --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,7 @@ +import sys +import os + +# Append project root directory to sys.path to import modules from src +sys.path.insert(0, os.path.abspath( + os.path.join(os.path.dirname(__file__), '..') +)) diff --git a/test_main.py b/tests/test_main.py similarity index 71% rename from test_main.py rename to tests/test_main.py index 192a049..3dafed1 100644 --- a/test_main.py +++ b/tests/test_main.py @@ -1,11 +1,34 @@ -import unittest import os +import uuid import json +import unittest from unittest.mock import patch + from typer.testing import CliRunner -from main import app -TEST_DRAFTS_FILE = "test-drafts.json" +from src.draft_utils import ensure_drafts_file + +# Import the 'app' object from the local module in the 'src' package. +from src.app import app + +# --- Alternative Method to Modify sys.path for Testing --- +# This approach is used to allow importing main.py, which is located in the project's root directory, +# when such tests rely on running the application entry point. +# Note: main.py defines a main() function and runs it if executed directly by the user. + +# # Uncomment the line below to update sys.path, adding the project root directory. +# import sys +# sys.path.insert(0, os.path.abspath( +# os.path.join(os.path.dirname(__file__), '..') +# )) +# from main import app + +# Generate a unique test drafts file name +TEST_DRAFTS_FILE = f"test-drafts-{uuid.uuid4().hex[:6]}.json" + +# Ensure TEST_DRAFTS_FILE path is resolved and the file exists, following XDG Base Directory Specification if necessary. +TEST_DRAFTS_FILE = ensure_drafts_file(TEST_DRAFTS_FILE) +os.remove(TEST_DRAFTS_FILE) class TestThreadsCLI(unittest.TestCase): def setUp(self): @@ -26,7 +49,7 @@ def test_get_recent_posts(self): self.assertIn("Recent Posts", result.stdout) def test_create_text_post(self): - with patch("main.create_post") as mock_create_post: + with patch("src.app.create_post") as mock_create_post: mock_create_post.return_value = {"id": "123"} result = self.runner.invoke(app, ["create-text-post", "Test post"]) self.assertEqual(result.exit_code, 0) @@ -56,7 +79,7 @@ def test_send_draft(self): draft_id = drafts[0]["id"] # Test sending an existing draft - with patch("main.create_text_post") as mock_create_text_post: + with patch("src.app.create_text_post") as mock_create_text_post: result = self.runner.invoke(app, ["send-draft", str(draft_id), "--drafts-file", TEST_DRAFTS_FILE]) self.assertEqual(result.exit_code, 0) self.assertIn(f"Draft with ID {draft_id} sent and removed from drafts.", result.stdout) @@ -70,4 +93,4 @@ def test_send_draft(self): self.assertIn("Draft with ID 999 not found.", result.stdout) if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main()