linkedin/main.py

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}