The instance where I was using it changed it's rules to prevent bots from posting in it and I didn't care enough to search for another instance.
https://lemm.ee/c/issue_tracker?dataType=Post&page=1&sort=Active
`config_template.py
LEMMY_INSTANCE_URL = ""
LEMMY_COMMUNITY_NAME = ""
LEMMY_USERNAME = ""
LEMMY_PASSWORD = ""
GITHUB_API_BASE = "https://api.github.com"
GITHUB_URL_BASE = "https://github.com"
REPOSITORIES = ["LemmyNet/lemmy", "LemmyNet/lemmy-ui"]
DB_FILE = "lemmy_github.db"
DELAY = 1
MAX_BACKOFF_TIME = 300
PERSONAL_ACCESS_TOKEN = ""
github_lemmy_issue_reposter.py
import backoff
import datetime
import logging
import requests
import schedule
import sqlite3
import time
from config import *
from pythorhead import Lemmy
from typing import Any, Dict, Generator, List, Optional, Tuple, Callable, TypeVar
T = TypeVar('T')
# "[%(levelname)s]:%(asctime)s:%(name)s [%(filename)s:%(lineno)s - %(funcName)s()] %(message)s"
FORMAT = "%(message)s"
logging.basicConfig(
level=logging.INFO,
format=FORMAT,
handlers=[logging.FileHandler("debug.log", mode="w"), logging.StreamHandler()],
)
def on_giveup(details: Dict[str, int]) -> None:
logging.error(f"Failed to fetch issues after {details['tries']} attempts", exc_info=True)
def handle_errors(message: Optional[str] = None) -> Callable[[Callable[..., T]], Callable[..., T]]:
def decorator(function: Callable[..., T]) -> Callable[..., T]:
def wrapper(*args: Tuple[Any], **kwargs: Dict[str, Any]) -> T:
try:
return function(*args, **kwargs)
except Exception as e:
if message:
logging.exception(f"{message} - Error in {function.__name__}:\n{e}")
else:
logging.exception(f"Error in {function.__name__}:\n{e}")
raise
return wrapper
return decorator
class GitHubIssue:
def __init__(self, issue_dict: dict[str, Any], github_repo: str) -> None:
try:
self.url = issue_dict["html_url"]
logging.info(f"Creating issue {self.url}")
self.state = issue_dict["state"]
self.state_fmt = "[Closed]" if issue_dict["state"] == "closed" else ""
self.repo_abbr = "[UI]" if "lemmy-ui" in github_repo else "[BE]"
self.title = f"{self.state_fmt}{self.repo_abbr} {issue_dict['title']} #{issue_dict['number']}"
self.title = self.title[:200]
self.body = issue_dict["body"]
if self.body is not None:
self.body = self.body[:30000]
self.user = issue_dict["user"]["login"]
self.user_url = issue_dict["user"]["html_url"]
self.updated_at = datetime.datetime.strptime(issue_dict["updated_at"], '%Y-%m-%dT%H:%M:%SZ')
except Exception as e:
log_message: str = (
f"Formatted issue:\n"
f" - Repo: {github_repo}\n"
f" - Issue State: {self.state}\n"
f" - Repo Abbreviation: {self.repo_abbr}\n"
f" - Title: {self.title}\n"
f" - URL: {self.url}\n"
f" - User: {self.user}\n"
f" - User URL: {self.user_url}\n"
f" - Updated At: {self.updated_at}\n"
)
logging.exception(log_message)
logging.exception(e)
@property
def formatted_body(self) -> str:
formatted_body: str = self.body
try:
if self.body is not None:
formatted_body = self.body.replace("\n", "\n> ")
formatted_body = f"> {formatted_body}\n> \n> *Originally posted by [{self.user}]({self.user_url}) in [#{self.number}]({self.url})*"
except Exception as e:
logging.exception(f"Error formatting body for {self.url}\n{e}")
return formatted_body
@property
def number(self) -> int:
return int(self.url.split("/")[-1])
class GitHubComment:
def __init__(self, comment_dict: dict[str, Any], issue_number: int) -> None:
self.id = comment_dict["id"]
self.body = comment_dict["body"]
self.user = comment_dict["user"]["login"]
self.user_url = comment_dict["user"]["html_url"]
self.url = comment_dict["html_url"]
self.issue_number = issue_number
@property
def formatted_comment(self) -> str:
formatted_body:str = self.body.replace("\n", "\n> ")
formatted_body = f"> {formatted_body}\n> \n> *Originally posted by [{self.user}]({self.user_url}) in [#{self.issue_number}]({self.url})*"
return formatted_body
@handle_errors("Error initializing database")
def initialize_database() -> sqlite3.Connection:
logging.info("Initializing database")
conn: sqlite3.Connection = sqlite3.connect(DB_FILE)
cursor: sqlite3.Cursor = conn.cursor()
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS posts (
issue_number INTEGER PRIMARY KEY,
lemmy_post_id INTEGER NOT NULL UNIQUE,
issue_title TEXT,
issue_body TEXT,
updated_at TIMESTAMP DEFAULT NULL
)
"""
)
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS comments (
github_comment_id INTEGER PRIMARY KEY,
lemmy_comment_id INTEGER NOT NULL UNIQUE,
comment_user TEXT,
comment_body TEXT
updated_at TIMESTAMP DEFAULT NULL
)
"""
)
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS last_updated (
id INTEGER PRIMARY KEY,
last_updated_time TIMESTAMP
);
"""
)
conn.commit()
return conn
def get_last_updated_time() -> str:
conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()
cursor.execute("SELECT last_updated_time FROM last_updated WHERE id = 1")
last_updated_time: str = cursor.fetchone()[0]
conn.close()
return last_updated_time
def update_last_updated_time() -> None:
conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()
current_time = datetime.datetime.utcnow().isoformat()
cursor.execute("UPDATE last_updated SET last_updated_time = ? WHERE id = 1", (current_time,))
if cursor.rowcount == 0:
cursor.execute("INSERT INTO last_updated (id, last_updated_time) VALUES (1, ?)", (current_time,))
conn.commit()
conn.close()
logging.info("Updated last updated time")
def update_post_time(post_id: int, updated_at: datetime.datetime) -> None:
conn: sqlite3.Connection = sqlite3.connect(DB_FILE)
cursor: sqlite3.Cursor = conn.cursor()
time_formatted = updated_at.strftime('%Y-%m-%d %H:%M:%S')
SQL = "UPDATE posts SET updated_at = ? WHERE lemmy_post_id = ?"
cursor.execute(SQL, (time_formatted, post_id))
conn.commit()
conn.close()
def check_updated_at(issue_number: int) -> Optional[Tuple[int, str, str, Optional[str]]]:
logging.info(f"Checking last post update for {issue_number}")
conn: sqlite3.Connection = sqlite3.connect(DB_FILE)
cursor: sqlite3.Cursor = conn.cursor()
SQL = "SELECT lemmy_post_id, issue_title, issue_body, updated_at FROM posts WHERE issue_number = ?"
cursor.execute(SQL, (issue_number,))
result: Tuple[int, str, str, Optional[str]] = cursor.fetchone()
conn.close()
if result is None:
logging.info(f"No post found for {issue_number}")
return None
else:
logging.info(f"Found post for {issue_number}")
return result
@handle_errors("Error initializing Lemmy instance")
def initialize_lemmy_instance() -> Lemmy:
logging.info("Initializing Lemmy instance")
lemmy = Lemmy(LEMMY_INSTANCE_URL)
logging.info(f"Initialized Lemmy instance in {LEMMY_INSTANCE_URL}")
lemmy.log_in(LEMMY_USERNAME, LEMMY_PASSWORD)
logging.info(f"Logged in to Lemmy instance with user {LEMMY_USERNAME}")
return lemmy
@backoff.on_exception(
backoff.expo,
(requests.exceptions.RequestException, TypeError),
max_time=MAX_BACKOFF_TIME,
on_giveup=on_giveup,
)
def fetch_github_data(url: str) -> List[Dict[str, Any]]:
global LAST_REQUEST_TIME
try:
headers = {
"Accept": "application/vnd.github+json",
"Authorization": f"Bearer {PERSONAL_ACCESS_TOKEN}",
"X-GitHub-Api-Version": "2022-11-28",
}
time_elapsed = time.time() - LAST_REQUEST_TIME
required_delay = max(0, DELAY - time_elapsed)
time.sleep(required_delay)
response = requests.get(url, headers=headers)
LAST_REQUEST_TIME = time.time()
logging.info(f"Fetched data from {url}")
res: List[Dict[str, Any]] = response.json()
return res
except requests.exceptions.RequestException as e:
logging.exception(f"Error fetching data from {url}\n{e}")
raise
def check_existing_post(issue_number: str) -> Optional[int]:
conn: sqlite3.Connection = sqlite3.connect(DB_FILE)
cursor: sqlite3.Cursor = conn.cursor()
SQL = "SELECT lemmy_post_id FROM posts WHERE issue_number=?"
cursor.execute(SQL, (issue_number,))
post_id: Optional[tuple[int]] = cursor.fetchone()
if post_id:
return post_id[0]
return None
def insert_post_to_db(issue: GitHubIssue, lemmy_post_id: Optional[int]) -> None:
try:
conn: sqlite3.Connection = sqlite3.connect(DB_FILE)
cursor: sqlite3.Cursor = conn.cursor()
SQL = "INSERT INTO posts (issue_number, lemmy_post_id, issue_title, issue_body, updated_at) VALUES (?, ?, ?, ?, ?)"
cursor.execute(SQL, (issue.number, lemmy_post_id, issue.title, issue.formatted_body, issue.updated_at))
conn.commit()
logging.info(f"Inserted new Lemmy post {lemmy_post_id} into the database")
except sqlite3.Error as e:
logging.exception(f"Error inserting post into the database for issue {issue.title} with url {issue.url}\n{e}")
raise
def insert_comment_to_database(cursor: sqlite3.Cursor, github_comment_id: int, lemmy_comment_id: int, comment: GitHubComment) -> None:
try:
SQL = "INSERT INTO comments (github_comment_id, lemmy_comment_id, comment_user, comment_body) VALUES (?, ?, ?, ?)"
cursor.execute(SQL, (github_comment_id, lemmy_comment_id, comment.user, comment.formatted_comment,))
logging.info(f"Inserted comment {github_comment_id} into the database")
except Exception as e:
logging.exception(f"Error encountered while inserting comment {github_comment_id} to database\n{e}")
@backoff.on_exception(
backoff.expo,
(requests.exceptions.RequestException, TypeError),
max_time=MAX_BACKOFF_TIME,
on_giveup=on_giveup,
)
def create_lemmy_post(lemmy: Any, community_id: int, issue: GitHubIssue) -> Optional[int]:
lemmy_post_id: Optional[int] = None
lemmy_post_id = lemmy.post.create(community_id, issue.title, url=issue.url, body=issue.body)["post_view"]["post"]["id"]
lemmy_url = f"{LEMMY_INSTANCE_URL}/post/{lemmy_post_id}"
logging.info(f"Posted issue {lemmy_url}")
return lemmy_post_id
@backoff.on_exception(
backoff.expo,
(requests.exceptions.RequestException, TypeError),
max_time=MAX_BACKOFF_TIME,
on_giveup=on_giveup,
)
def create_lemmy_comment(lemmy: Any, post_id: Optional[int], comment: GitHubComment) -> Optional[int]:
logging.info(f"Creating new Lemmy comment in {LEMMY_INSTANCE_URL}/post/{post_id}")
if not post_id:
logging.warning("Post ID is empty. Skipping comment creation")
return None
response = lemmy.comment.create(post_id, comment.formatted_comment)
lemmy_comment_id:int = response["comment_view"]["comment"]["id"]
logging.info(f"Successfully created Lemmy comment {LEMMY_INSTANCE_URL}/comment/{lemmy_comment_id}")
return lemmy_comment_id
def get_total_issues(github_repo: str) -> int:
url: str = f"https://api.github.com/repos/{github_repo}"
data: List[Dict[str, Any]] = fetch_github_data(url)
total_issues: int = data["open_issues_count"]
return total_issues
def fetch_issues(github_repo: str, last_updated_time: str) -> Generator[Dict[str, Any], None, None]:
page = 1
per_page = 100
issues_url = (f"{GITHUB_API_BASE}/repos/{github_repo}/issues?state=all&since={last_updated_time}&per_page={per_page}")
while True:
page_url = f"{issues_url}&page={page}"
issues: List[Dict[str, Any]] = fetch_github_data(page_url)
if not issues:
break
for issue_dict in issues:
yield issue_dict
page += 1
@backoff.on_exception(
backoff.expo,
(requests.exceptions.RequestException, TypeError),
max_time=MAX_BACKOFF_TIME,
on_giveup=on_giveup,
)
def edit_lemmy_post(lemmy: Any, lemmy_post_id: int, issue: GitHubIssue) -> None:
lemmy.post.edit(lemmy_post_id, name=issue.title, url=issue.url, body=issue.body)
def process_issues(lemmy: Any, community_id: int, github_repo: str) -> None:
last_updated_time = get_last_updated_time()
update_last_updated_time()
for issue_dict in fetch_issues(github_repo, last_updated_time):
process_issue(lemmy, community_id, github_repo, issue_dict)
def process_issue(lemmy: Any, community_id: int, github_repo: str, issue_dict: dict[str, Any]) -> None:
issue: GitHubIssue = GitHubIssue(issue_dict, github_repo)
res: Optional[Tuple[int, str, str, Optional[str]]] = check_updated_at(issue.number)
if res is None:
create_new_lemmy_post(lemmy, community_id, github_repo, issue)
else:
lemmy_post_id, existing_title, existing_body, updated_at = res
if updated_at is None or has_enough_time_passed(updated_at, issue.updated_at):
update_issue_if_needed(lemmy, lemmy_post_id, existing_title, existing_body, issue)
process_comments(lemmy, lemmy_post_id, github_repo, issue)
update_post_time(lemmy_post_id, issue.updated_at)
def has_enough_time_passed(old_updated_at_str: str, new_updated_at: datetime.datetime) -> bool:
old_updated_at = datetime.datetime.strptime(old_updated_at_str, '%Y-%m-%d %H:%M:%S')
time_difference: datetime.timedelta = new_updated_at - old_updated_at
return time_difference >= datetime.timedelta(hours=2)
def update_issue_if_needed(lemmy: Any, lemmy_post_id: int, existing_title: str, existing_body: str, issue: GitHubIssue) -> None:
if existing_title != issue.title or existing_body != issue.formatted_body:
edit_lemmy_post(lemmy, lemmy_post_id, issue)
def create_new_lemmy_post(lemmy: Any, community_id: int, github_repo: str, issue: GitHubIssue) -> None:
lemmy_post_id: Optional[int] = post_issue_to_lemmy(lemmy, community_id, issue)
insert_post_to_db(issue, lemmy_post_id)
process_comments(lemmy, lemmy_post_id, github_repo, issue)
def post_issue_to_lemmy(lemmy: Any, community_id: int, issue: GitHubIssue) -> Optional[int]:
try:
logging.info(f"Start posting issue {issue.title} to community {community_id}")
lemmy_post_id: Optional[int] = create_lemmy_post(lemmy, community_id, issue)
return lemmy_post_id
except Exception as e:
logging.exception(f"Error posting issue {issue.title} to community {community_id}\n{e}")
return None
def process_comments(lemmy: Any, post_id: Optional[int], github_repo: str, issue: GitHubIssue) -> None:
try:
logging.info(f"Posting comments from issue #{issue.number} to Lemmy post {LEMMY_INSTANCE_URL}/post/{post_id}")
comments_url: str = f"{GITHUB_API_BASE}/repos/{github_repo}/issues/{issue.number}/comments"
comments: Dict[str, Any] = fetch_github_data(comments_url)
for comment_data in comments:
if isinstance(comment_data, str):
logging.warning(f"Skipping comment {comment_data}")
continue
process_comment(lemmy, github_repo, comment_data, post_id, issue.number)
except Exception as e:
logging.exception(f"Error posting comments to lemmy post {post_id}\n{e}")
def process_comment(lemmy: Any, github_repo: str, comment_data: Dict[str, Any], post_id: Optional[int], issue_number: int) -> None:
conn: sqlite3.Connection = sqlite3.connect(DB_FILE)
cursor: sqlite3.Cursor = conn.cursor()
comment = GitHubComment(comment_data, issue_number)
existing_comment_id: Optional[int] = get_existing_comment_id(cursor, comment.id)
if existing_comment_id:
logging.info(f"Skipping existing comment with GitHub comment ID: {comment.id}")
return
post_comment_to_lemmy(cursor, lemmy, github_repo, comment, post_id, issue_number)
conn.commit()
def post_comment_to_lemmy(cursor: sqlite3.Cursor, lemmy: Any, github_repo: str, comment: GitHubComment, post_id: Optional[int], issue_number: int) -> None:
lemmy_post_url = f"{LEMMY_INSTANCE_URL}/post/{post_id}"
comment_url = f"{GITHUB_URL_BASE}/{github_repo}/issues/{issue_number}#issuecomment-{comment.id}"
logging.info(f"Posting comment {comment.url} to Lemmy post {lemmy_post_url}")
lemmy_comment_id: Optional[int] = create_lemmy_comment(lemmy, post_id, comment)
if not lemmy_comment_id:
logging.exception(f"Error creating Lemmy comment {lemmy_comment_id} to {lemmy_post_url} from Github comment {comment.url}")
return
logging.info(f"Posted comment {comment_url} to Lemmy post {lemmy_post_url}")
insert_comment_to_database(cursor, comment.id, lemmy_comment_id, comment)
def get_existing_comment_id(cursor: sqlite3.Cursor, github_comment_id: int) -> Optional[int]:
logging.info(f"Checking if comment with GitHub comment ID: {github_comment_id} exists")
cursor.execute("SELECT lemmy_comment_id FROM comments WHERE github_comment_id=?", (github_comment_id,))
existing_comment = cursor.fetchone()
if existing_comment is not None:
logging.info(f"Found existing comment with GitHub comment ID: {github_comment_id}")
existing_comment_id: int = existing_comment[0]
return existing_comment_id
else:
logging.info(f"No existing comment found with GitHub comment ID: {github_comment_id}")
return None
def fetch_issue_data(github_repo: str) -> List[Tuple[str, Optional[int]]]:
logging.info("Fetching the GitHub issue number and Lemmy post ID for all issues")
conn: sqlite3.Connection = sqlite3.connect(DB_FILE)
cursor: sqlite3.Cursor = conn.cursor()
SQL = "SELECT issue_url, lemmy_post_id FROM posts WHERE issue_url LIKE ?"
issues_url = f"https://github.com/{github_repo}/issues/%"
issue_data = cursor.execute(SQL, (issues_url,)).fetchall()
logging.info(f"Fetched {len(issue_data)} issues")
return issue_data
def process_repo(lemmy: Any, community_id: int, github_repo: str) -> None:
try:
logging.info(f"Processing repository {github_repo}")
process_issues(lemmy, community_id, github_repo)
except Exception as e:
logging.exception(f"Error occurred while processing repository {github_repo}\n{e}")
def main() -> None:
logging.info("Running main function")
initialize_database()
lemmy = initialize_lemmy_instance()
community_id = lemmy.discover_community(LEMMY_COMMUNITY_NAME)
for github_repo in REPOSITORIES:
process_repo(lemmy, community_id, github_repo)
def run_periodically() -> None:
logging.info("Starting periodic run")
schedule.every(1).hours.do(main)
while True:
try:
schedule.run_pending()
except Exception as e:
logging.exception(f"Error occurred during scheduling\n{e}")
time.sleep(60)
if __name__ == "__main__":
try:
logging.info("Starting script")
main()
run_periodically()
except Exception as e:
logging.exception(f"Error occurred during script execution\n{e}")
requirements.txt
pythorhead==0.12.3
schedule==1.2.0
backoff==2.2.1
feedparser==6.0.10
