Job Summary :
The Client Technologies Specialist 1, also known as an Inventory Management Specialist, is a full-time role focused on managing inventory operations for the Office of Information Systems. The specialist will oversee inventory based on the needs of network services, perform regular audits, maintain accurate reports, forecast inventory needs, and coordinate logistics. This position requires strong organizational, analytical, and data entry skills, as well as familiarity with inventory management systems.
Location : Augusta, Maine, United States
Washington, District of Columbia, United States
Responsibilities :
Required Skills & Certifications :
#src / main.py
import os
import json
import re
import PyPDF2
import docx
import requests
import pdfplumber
import docx
import hashlib
import io
import zipfile
import xml.etree.ElementTree as ET
import openpyxl
import hashlib
import unicodedata
import regex
from typing import Pattern
from flask import Flask, jsonify, request
from google.cloud import firestore
from flask_cors import CORS
from datetime import datetime, timezone
from dotenv import load_dotenv
import firebase_admin
from firebase_admin import credentials, firestore, storage
from bs4 import BeautifulSoup # for URL text extraction
import PyPDF2
import unicodedata, regex, uuid
from crew_ai.document_crew import run_analysis
import asyncio
from concurrent.futures import ThreadPoolExecutor
import uuid
from urllib.parse import unquote
from threading import Thread
from itertools import zip_longest
from crewai import Crew
from crew_ai.agent import capture
from crew_ai.document_task import capture_task
from crew_ai.document_task import extract_orgname # adjust import if needed
import regex
import unicodedata
from typing import Union
import unicodedata
# Load environment variables
load_dotenv()
# Initialize Flask app
app = Flask(__name__) # MUST come first
# Configure CORS
allowed_origins = os.getenv("ALLOWED_ORIGINS", "
CORS(app, resources={r" /
service_account_info = json.loads(os.getenv("FIREBASE_SERVICE_ACCOUNT"))
cred = credentials.Certificate(service_account_info)
firebase_admin.initialize_app(cred, {
"storageBucket" : service_account_info.get("storageBucket")
})
db = firestore.client()
bucket = storage.bucket()
print(f" Firebase initialized with bucket : {bucket.name}")
# Azure OpenAI Setup
AZURE_API_KEY = os.getenv("AZURE_OPENAI_API_KEY")
AZURE_API_BASE = os.getenv("AZURE_OPENAI_API_BASE")
AZURE_API_VERSION = os.getenv("AZURE_OPENAI_API_VERSION")
AZURE_DEPLOYMENT_NAME = os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME")
AZURE_API_URL = f"{AZURE_API_BASE} / openai / deployments / {AZURE_DEPLOYMENT_NAME} / chat / completions?api-version={AZURE_API_VERSION}"
def extract_pdf_text(content) :
text = ""
try :
with pdfplumber.open(io.BytesIO(content)) as pdf :
for page in pdf.pages :
page_text = page.extract_text()
if page_text :
text += page_text + "
except Exception as e :
print(" PDF parsing error : " + str(e))
return text
def extract_docx_text(content) :
text = ""
try :
doc = docx.Document(io.BytesIO(content))
text = "
".join([p.text for p in doc.paragraphs])
for table in doc.tables :
for row in table.rows :
text += "
" + " ".join(cell.text for cell in row.cells)
except Exception as e :
print(" DOCX parsing error : " + str(e))
return text
def extract_txt_text(content) :
try :
return content.decode("utf-8", errors="ignore")
except Exception as e :
print(" TXT decoding error : " + str(e))
return ""
def extract_text_by_type(filename, content) :
filename_lower = filename.lower()
if filename_lower.endswith(".pdf") :
return extract_pdf_text(content)
if filename_lower.endswith(".docx") :
return extract_docx_text(content)
if filename_lower.endswith(".txt") :
return extract_txt_text(content)
if filename_lower.endswith(".pptx") :
return extract_pptx_text(content)
if filename_lower.endswith(".xlsx") :
return extract_xlsx_text(content)
print(f" Unsupported file type : {filename}")
return ""
def get_files_from_firebase_folder(folder_name) :
"""
Reads all files from Firebase Storage folder, extracts text, and returns as a single string.
"""
texts = []
try :
blobs = list(bucket.list_blobs(prefix=f"Upload / {folder_name} / "))
print(f"
Client Specialist • Augusta, ME, United States