helpers.py

from datetime import timedelta, datetime
from collections import defaultdict
import json
from typing import Tuple
from google.cloud.storage import Client, transfer_manager

from config import JOB_TYPES, PERSON_NAME, PERSON_PRONOUN, RESET_QUERY, CV_DATA_DIRNAME, CV_DATA_GCS_BUCKET


def download_cv_data() -> None:
    # download cv data from google cloud storage
    storage_client = Client()
    bucket = storage_client.bucket(CV_DATA_GCS_BUCKET)
    cv_document_blobs = [blob.name for blob in bucket.list_blobs(prefix=CV_DATA_DIRNAME)]
    results = transfer_manager.download_many_to_path(bucket, cv_document_blobs)
    if not all(result is None for result in results):
        exception_message = ""
        for i, download_result in enumerate(results):
            if download_result is not None:
                exception_message += f"❗ Exception occurred while downloading {cv_document_blobs[i]} from {CV_DATA_GCS_BUCKET}: {download_result}; "
        
        raise RuntimeError(exception_message)

def get_job_type(position: str) -> str:
    position_lower = position.lower()
    for job_type in JOB_TYPES:
        if job_type.lower() in position_lower:
            return job_type
    raise ValueError(f"Position '{position}' does not match any known job types.")

def join_grammatically(words: list[str]) -> str:
    if len(words) == 0:
        raise ValueError("words must contain at least one word.")
    elif len(words) == 1:
        return words[0]
    return ", ".join(words[:-2] + [" and ".join(words[-2:])])

def td_to_yoe_str(td: timedelta) -> str:
    secs_in_year = 365.2425 * 24 * 3600
    td_secs = td.total_seconds()
    if td_secs < secs_in_year:
        secs_in_month = 30.436875 * 24 * 3600
        months = round(td_secs / secs_in_month)
        return f"{months} {'month' if months <= 1 else 'months'}"
    else:
        return f"{td_secs / secs_in_year:.1f} years"

def get_job_documents(jobs_json: dict) -> Tuple[str, list[str]]:
    jobs_docs = []
    positions_in_order = []
    yoe_by_job_type = defaultdict(timedelta)
    for job_json in jobs_json["jobs"].values():
        company_word = f"[{job_json["company"]["name"]}]({job_json["company"]["link"]}) in {job_json["company"]["location"]}"
        for position in job_json["positions"]:
            start_date = datetime.strptime(position["startDate"], "%Y-%m-%d")
            end_date = datetime.strptime(position["endDate"], "%Y-%m-%d")
            yoe_by_job_type[get_job_type(position["name"])] += end_date - start_date

            job_doc = f"{PERSON_NAME} worked as {position["name"]} at {company_word} from {start_date.strftime('%B %Y')} to {end_date.strftime('%B %Y')}."
            job_doc += f" Following are highlights of this position: {'; '.join(position['highlights'])}."
            jobs_docs.append(job_doc)

            positions_in_order.append(f"➢ {position["name"]} at {company_word} ({start_date.strftime('%B %Y')} - {end_date.strftime('%B %Y')})")
    
    yoe_by_job_type_strs = [f"{td_to_yoe_str(yoe)} of experience as {job_type}" for job_type, yoe in yoe_by_job_type.items()]
    intro_message = f"{PERSON_NAME} has {join_grammatically(yoe_by_job_type_strs)}. Following are positions {PERSON_PRONOUN} has worked as:\n"
    intro_message += "\n".join(positions_in_order)
    intro_message += f"\nYou can ask me about more details on any of these positions. As a reminder, you can say \'{RESET_QUERY}\' to reset chat."

    return intro_message, jobs_docs

def get_education_documents(education_json: dict) -> Tuple[str, list[str]]:
    education_docs = []

    university_name = education_json["school"]["name"]
    university_end_date = datetime.strptime(education_json["endDate"], "%Y-%m-%d").strftime("%B %Y")
    degree_full_str = f"{education_json["degree"]["name"]}"
    minors = [minor["name"] for minor in education_json["minors"]]
    if len(minors) == 1:
        degree_full_str += f" with minor in {minors[0]}"
    elif len(minors) > 1:
        degree_full_str += f" with minors in {", ".join(minors)}"

    dept_courses = defaultdict(list[str])
    for course_json in education_json["courses"]:
        course_number = course_json["number"]
        course_dept = course_number.split(" ")[0]
        dept_courses[course_dept].append(f"➢ [{course_json["number"]}]({course_json["link"]}): {course_json["title"]}")

    dept_names = education_json["courseDepartments"]
    sorted_depts = sorted(dept_courses, key=lambda k: len(dept_courses[k]), reverse=True)
    for dept in sorted_depts:
        education_docs.append(f"In department of {dept_names[dept]}, {PERSON_NAME} took \n{"\n".join(dept_courses[dept])}.")
    sorted_dept_names = "\n".join([f"➢ {dept_names[dept]} ({dept})" for dept in sorted_depts])

    intro_message = (f"{PERSON_NAME} graduated from {university_name} in {university_end_date} with {degree_full_str}. "
                     f"{PERSON_PRONOUN.title()} took {len(education_json["courses"])} courses across following departments:\n{sorted_dept_names}."
                     f"\nYou can ask me about detailed list of courses {PERSON_NAME} took in each department. As a reminder, you can say \'{RESET_QUERY}\' to reset chat.")
    
    return intro_message, education_docs

def get_paper_documents(papers_json: dict) -> Tuple[str, list[str]]:
    intro_messages = []
    paper_docs = []
    for i, paper_json in enumerate(papers_json["papers"]):
        paper_date = datetime.strptime(paper_json["date"], "%Y-%m")
        intro_messages.append(f"{i + 1}: {paper_json["title"]} ({paper_date.strftime("%Y")})")
        paper_doc = f"{i + 1}: Title of this paper is {paper_json["title"]}."

        location_str = f" in {paper_json["location"]}" if "location" in paper_json else ""
        paper_doc += f" It was published on {paper_date.strftime("%B %Y")}{location_str}."

        if "conference" in paper_json:
            paper_doc += f" It was published in {paper_json["conference"]}."

        paper_doc += f" Link to paper is {paper_json["paperUrl"]}."

        paper_doc += f" Following is description of this paper. {paper_json["desc"]}"

        paper_docs.append(paper_doc)
    
    intro_message = (f"{PERSON_NAME} wrote {len(intro_messages)} papers. Here are titles of these papers:\n{"\n".join(intro_messages)}"
                     f"\nYou can ask me about more details on any of these papers. As a reminder, you can say \'{RESET_QUERY}\' to reset chat.")
    
    return intro_message, paper_docs

def get_skill_documents(skills_json: dict) -> Tuple[str, list[str]]:
    skill_competency = defaultdict(int)
    skills_by_category = defaultdict(list)
    for skill_json in skills_json["skills"]:
        skill_competency[skill_json["title"]] = skill_json["competency"]
        for skill_category in skill_json["category"]:
            skills_by_category[skill_category].append(skill_json["title"])
    
    skill_docs = []
    for category, skills in skills_by_category.items():
        skills.sort(key=lambda x: skill_competency[x], reverse=True)
        skill_docs.append(f"Technical skills related to {category} that {PERSON_NAME} has are: {", ".join(skills)}.")

    sorted_skill_categories = list(skills_by_category.keys())
    sorted_skill_categories.sort()
    intro_message = (f"{PERSON_NAME} has various technical skills across following categories:\n"
                     f"{'\n'.join([f"➢ {category}" for category in sorted_skill_categories])}"
                     f"\nYou can ask me which technical skills {PERSON_NAME} has in one of these areas. As a reminder, you can say \'{RESET_QUERY}\' to reset chat.")

    return intro_message, skill_docs

def print_docs(name, file_name, parse_func):
    with open(f"cvdata/{file_name}") as f:
        intro_message, docs = parse_func(json.load(f))

    print(f'\n>>>>>>>>>> {name} <<<<<<<<<<')
    print('========== Intro')
    print(intro_message)
    print('\n========== Docs')
    for doc in docs:
        print(doc + '\n')


if __name__ == "__main__":
    print_docs("Jobs", "jobs.json", get_job_documents)
    print_docs("Education", "education.json", get_education_documents)
    print_docs("Papers", "papers.json", get_paper_documents)
    print_docs("Skills", "skills.json", get_skill_documents)