Spaces:
Running on Zero
Running on Zero
| 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) |