linkedin/app/ones.py

import sys
import os

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

from linkedin_api import Linkedin
import logging
import schedule
import time
from utils.logging_setup import configure_global_logging
from datetime import datetime
import json

from dotenv import load_dotenv
import os

configure_global_logging()
load_dotenv()

# # Ваши учетные данные LinkedIn
username = os.getenv('USERNAME')
password = os.getenv('PASSWD')


# Authenticate using any Linkedin user account credentials
api = Linkedin(username, password)

def pars_jobs(geo):
    search_jobs = api.search_jobs(location_geo_id = geo)
    search_jobes = [
        {
        "title": item.get("title"),
        "entityUrn": item.get("entityUrn"),
               
        
        }
    for item in search_jobs
    ]
    # print(search_job)
    file_path = "search_jobes.json"
    with open(file_path, "w", encoding="utf-8") as json_file:
        json.dump(search_jobes, json_file, indent=4, ensure_ascii=False)

    print(f"Результаты успешно сохранены в {file_path}")
    
    
def add_to_bd():
    #[ ]: Написать функцию записи в БД
    pass
    
def get_job(job_id):
    jobs = api.get_job(job_id)
    text = jobs['description']['text']
    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)
    jobPostingId = jobs['jobPostingId']
    localizedName = jobs['workplaceTypesResolutionResults']
    # Извлекаем все localizedName
    localized_names = [value['localizedName'] for value in localizedName.values()]
    localized_name = ", ".join(localized_names)
    # localized_name = workplaceTypesResolutionResults['urn:li:fs_workplaceType:2']['localizedName']
    link = f'https://www.linkedin.com/jobs/view/{job_id}/'
    names = jobs['companyDetails']['com.linkedin.voyager.deco.jobs.web.shared.WebCompactJobPostingCompany']['companyResolutionResult']['name']
    url = jobs['companyDetails']['com.linkedin.voyager.deco.jobs.web.shared.WebCompactJobPostingCompany']['companyResolutionResult']['url']
    
    # [ ]: job_level job_type hourly_rate найти minimum_annual_salary и salary_currency добавил  description  names Компании url на компанию
    logging.info(f"title: {title},  location: {location},  jobPostingId: {jobPostingId}, difference: {difference}, location_type: {localized_name}, link: {link}  ===== {names} {url}") #text:{text},

    




def get_vakansi():
    # Чтение данных из JSON-файла
    file_path = "search_jobes.json"
    with open(file_path, "r", encoding="utf-8") as json_file:
        data = json.load(json_file)

    # Вывод данных
    for d in data:
        title = d['title']
        entityUrn = d['entityUrn']
        job_id = entityUrn.split(":")[-1]
        get_job(job_id)
  
        print(f"{title} {job_id}")
        
        




geo = '100025096'
# pars_jobs(geo)




def main():
    # get_vakansi()   
    # logging.info("WORK") 
    get_job('4130181356')
    
    
    
    
    
main()
    
# [x] Обмен по времени не удалять main() что бы при старте сразу отрабатывала)
# Запуск функции каждые 5 минут 
# schedule.every(5).minutes.do(main)


# Запуск функции каждые день
# schedule.every().day.at("08:30").do(main)


# # Основной цикл выполнения
# while True:
#     schedule.run_pending()  # Запускает запланированные задачи
    
#     time.sleep(1)  # Пауза между проверками