This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import csv | |
| import datetime | |
| import getpass | |
| import requests | |
| urls = {} | |
| urls["host"] = "https://histre.com/" | |
| urls["api"] = f"{urls['host']}api/v1/" | |
| urls["auth"] = f"{urls['api']}auth_token/" | |
| urls["collection"] = f"{urls['api']}collections/" | |
| urls["note"] = f"{urls['api']}note/?append=true" | |
| csv_path = "/home/kiru/Downloads/director-of-product-results.csv" | |
| def authenticate(username=None, password=None): | |
| if username is None: | |
| username = input("Enter your histre username or email: ") | |
| if password is None: | |
| password = getpass.getpass("Enter your histre password: ") | |
| json = { | |
| "username": username, | |
| "password": password, | |
| } | |
| r = requests.post(urls["auth"], json=json) | |
| if r.status_code != 200: | |
| print(r.text) | |
| raise Exception("Authentication failed") | |
| response = r.json() | |
| auth_token = response["data"]["access"] | |
| return auth_token | |
| def auth_header(auth_token): | |
| return {"Authorization": f"Bearer {auth_token}"} | |
| def create_collection(title, description, headers): | |
| json = {"title": title, "description": description} | |
| r = requests.post(urls["collection"], json=json, headers=headers) | |
| if r.status_code >= 400: | |
| print(r.text) | |
| raise Exception("Collection creation failed") | |
| response = r.json() | |
| book_id = response["data"]["book_id"] | |
| return book_id | |
| def import_linkedin(): | |
| """Import LinkedIn results (exported as csv) into histre""" | |
| auth_token = authenticate() | |
| headers = auth_header(auth_token) | |
| # create collection | |
| title = "LinkedIn - Director of Product" | |
| utc_dt = datetime.datetime.now() | |
| desc = f"Uploaded at {utc_dt.astimezone().strftime('%c')}" | |
| book_id = create_collection(title, desc, headers) | |
| book_url = f"{urls['host']}collections/{book_id}/" | |
| print(f"Created collection {book_url}") | |
| # save notes to collection | |
| notes = [] | |
| with open(csv_path) as csvfile: | |
| reader = csv.reader(csvfile, delimiter=",") | |
| next(reader, None) # skip headers | |
| for sub in reader: | |
| if len(sub) < 5: | |
| continue | |
| url = sub[4] | |
| title = sub[0] | |
| note = f"<p>{sub[1]}<br />{sub[2]}<br />{sub[3]}</p>" | |
| json = { | |
| "url": url, | |
| "title": title, | |
| "note": note, | |
| "book_ids": [book_id], | |
| } | |
| notes.append(json) | |
| r = requests.post(urls["note"], json=notes, headers=headers) | |
| if r.status_code >= 400: | |
| print(r.text) | |
| raise Exception("Note creation failed") | |
| print("Successfully uploaded all results!") | |
| return | |
| if __name__ == "__main__": | |
| import_linkedin() |