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)