from fastapi import FastAPI, HTTPException from pydantic import BaseModel, HttpUrl from urllib.parse import urlparse, parse_qs from linkedin_api import Linkedin import logging from utils.logging_setup import configure_global_logging from datetime import datetime from dotenv import load_dotenv import os load_dotenv() # # Ваши учетные данные LinkedIn configure_global_logging() app = FastAPI( title="linkedin API", description="API с linkedin получаем данные по URL", version="1.0.0" ) username = os.getenv('USERNAME') password = os.getenv('PASSWD') api = Linkedin(username, password) class Profile(BaseModel): name: str title: str url: HttpUrl class UrlModel(BaseModel): link_profile: HttpUrl # Используем HttpUrl для проверки валидности URL @app.post("/link_profile") async def profile_url(data: UrlModel): # Проверяем, что ссылка начинается с "https://xander" if not str(data.link_profile).startswith("https://www.linkedin.com/"): # Возвращаем исключение с кодом 400 и сообщением об ошибке raise HTTPException(status_code=400, detail="Неправильная ссылка") url_profiles = str(data.link_profile) path = urlparse(url_profiles).path print(path) # Пропускаем '/in/' и получаем последнюю часть profiles = path.split("/")[2] print(profiles) profile = api.get_profile(profiles) logging.info(f"{profile}") firstName = profile['firstName'] lastName = profile['lastName'] locationName = profile['locationName'] geoLocationName = profile['geoLocationName'] headline = profile['headline'] educations = profile['education'] education = [ { "schoolName": item.get("schoolName"), "Field_of_Study": item.get("fieldOfStudy"), "Degree": item.get('degreeName'), "Grade": "??", "Start Date": item.get("timePeriod", {}).get("startDate", {}).get("year"), "End Date": item.get("timePeriod", {}).get("endDate", {}).get("year"), "Currently study here": '??', "Description": "??" } for item in educations ] experiences = profile['experience'] experience = [ { "companyName": item.get("companyName"), "title": item.get("title"), "Employment type": "??", "Location": item.get('geoLocationName'), "Location Type": "??", "Start Date": { "month": item.get("timePeriod", {}).get("startDate", {}).get("month"), "year": item.get("timePeriod", {}).get("startDate", {}).get("year"), }, "End Date":{ "month": item.get("timePeriod", {}).get("endDate", {}).get("month"), "year": item.get("timePeriod", {}).get("endDate", {}).get("year"), }, "Currently study here": '??', "Description": item.get('description') } for item in experiences ] volunteers = profile['volunteer'] volunteer = [ { "Organization": item.get("companyName"), "Role": item.get("role"), "Cause": item.get("cause"), "Location": item.get('geoLocationName'), "Start Date": { "month": item.get("timePeriod", {}).get("startDate", {}).get("month"), "year": item.get("timePeriod", {}).get("startDate", {}).get("year"), }, "End Date":{ "month": item.get("timePeriod", {}).get("endDate", {}).get("month"), "year": item.get("timePeriod", {}).get("endDate", {}).get("year"), }, "Currently study here": '??', "Description": item.get('description') } for item in volunteers ] all_skills = profile['skills'] skills = [item['name'] for item in all_skills] all_languages = profile['languages'] languages = [item['name'] for item in all_languages] # Если профиль не в виде словаря, возвращаем его напрямую return {"FirstName": firstName, "LastName":lastName, "Location": f"{geoLocationName} {locationName}", "Statement":headline, 'Education - list': education, 'Experience - list': experience, "Volunteering - list": volunteer, "skills": skills, "languages":languages} @app.post("/link_vacancy") async def vacancy_url(data: UrlModel): if not str(data.link_profile).startswith("https://www.linkedin.com/"): # Возвращаем исключение с кодом 400 и сообщением об ошибке raise HTTPException(status_code=400, detail="Неправильная ссылка") query_params = parse_qs(urlparse(str(data.link_profile)).query) current_job_id = query_params.get("currentJobId", [None])[0] jobs = api.get_job(current_job_id) location = jobs['formattedLocation'] title = jobs['title'] listedAt = jobs['listedAt'] # Преобразование из миллисекунд в секунды и конвертация future_date = datetime.fromtimestamp(listedAt / 1000) # Текущая дата current_date = datetime.now() # Разница в днях difference = abs((future_date - current_date).days) # print(f"Разница в днях: {difference}") jobPostingId = jobs['jobPostingId'] # workplaceTypesResolutionResults = jobs['workplaceTypesResolutionResults'] # Извлекаем все localizedName # localized_names = [value['localizedName'] for value in workplaceTypesResolutionResults.values()] # print(localized_names) # localized_name = workplaceTypesResolutionResults['urn:li:fs_workplaceType:2']['localizedName'] link = f'https://www.linkedin.com/jobs/view/{current_job_id}/' return {"job_id": current_job_id, "job_title": title, "minimum_annual_salary": f"minimum_annual_salary", "salary_currency": 'salary_currency', 'location_type': "location_type", 'location': location, "job_level": "job_level", "job_type": 'job_type', "days_posted": difference, "hourly_rate": "hourly_rate", "link": link}