app.py

from flask import Flask, render_template, request, redirect, url_for, session
from bs4 import BeautifulSoup
import requests
import boto3
from boto3.dynamodb.conditions import Attr
import random
import time

#爬蟲主程式檔案
app = Flask(__name__)
app.secret_key = 'your_secret_key' 

#################################
#            爬蟲               #
################################

# 工作內容
def job_content(soup):
    # 找到包含特定屬性值的 div 元素
    job_description_div = soup.find('div', {'class': 'job-address'})

    if job_description_div:
        # 獲取 job_description 屬性的值
        job_description_text = job_description_div.get('jobdescription', '')

        # 解析 HTML 格式的文字內容
        soup_description = BeautifulSoup(job_description_text, 'html.parser')
        text_content = soup_description.get_text(strip=True)

        return text_content

    # 如果沒有找到相應元素，返回空字符串
    return ''


def extract_data(soup, section_title):
    # 找到包含特定標題的h3標籤
    h3_element = soup.find('h3', {'class': 'h3'}, string=section_title)

    if h3_element:
        # 找到h3標籤的父級div
        div_element = h3_element.find_parent('div', {'class': 'list-row__head'})

        # 找到目標元素
        target_elements = div_element.find_next_sibling().find_all(['u', 'p'])

        # 獲取文字內容
        result_texts = [element.text for element in target_elements]
        return result_texts
    else:
        return []

#若不拘
def extract_data_general(soup, section_title):
    # 找到包含特定標題的h3標籤
    h3_element = soup.find('h3', {'class': 'h3'}, string=section_title)

    if h3_element:
        div_element = h3_element.find_parent('div', {'class': 'list-row__head'})
        target_element = div_element.find_next_sibling().find('div', {'class': 't3 mb-0'})
        result_text = target_element.text.strip()
        return [result_text]
    else:
        return []

def extract_data_based_on_tag(soup, section_title):
    result_texts = extract_data(soup, section_title)
    
    if not result_texts:
        result_texts = extract_data_general(soup, section_title)

    return result_texts

# 公司名稱
def get_company_name(soup):
    company_name_element = soup.find('a', {'class': 'btn-link', 'data-gtm-head': '公司名稱'})
    if company_name_element:
        return company_name_element['title']
    else:
        return ""

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/signup')
def signup():
    return render_template('signup.html')

@app.route('/submit', methods=['POST'])
def submit():
    user_input_url = request.form['url']

    #需修改成訊息方塊
    # 防呆-1 檢查是否為104的網址
    if not user_input_url.startswith('https://www.104.com.tw/job/'):
        return "錯誤：請輸入有效的 104 職缺網址（https://www.104.com.tw/job/ 開頭）"

    response = requests.get(user_input_url)

    if response.status_code == 200:
        soup = BeautifulSoup(response.text, 'html.parser')

        job_result_texts = extract_data_based_on_tag(soup, '職務類別') #tag存在
        tools_result_texts = extract_data_based_on_tag(soup, '擅長工具') #tag存在
        skills_result_texts = extract_data_based_on_tag(soup, '工作技能') #tag存在 但尚未有完整列表
        other_result_texts = extract_data_based_on_tag(soup, '其他條件')
        company_result_text = get_company_name(soup)
        job_content_text = job_content(soup)

        # 將資訊傳遞給模板
        return render_template('Job_info.html',
                               job_content_text=job_content_text,
                               job_result_texts=job_result_texts,
                               tools_result_texts=tools_result_texts,
                               skills_result_texts=skills_result_texts,
                               other_result_texts=other_result_texts,
                               company_result_text=company_result_text)

    else:
        return f"無法獲取網頁內容，狀態碼: {response.status_code}"

#################################################################################
#    確認職務資料後點選提交 交由系統依據職務類別+擅長工具比對資料庫關鍵字、分配題目   #
###############################################################################
    
# 初始化 DynamoDB client
dynamodb = boto3.resource('dynamodb', region_name='ap-east-1')
table_name = 'questions'
table = dynamodb.Table(table_name)

@app.route('/assign', methods=['POST'])
def assign():
    if request.method == 'POST':

        # 獲取表單提交的擅長工具文本 + 職務類別文本
        tool_text = request.form['tools']
        job_text = request.form['jobs']

        # 將擅長工具文本 + 職務類別文本進行分割，再對每個字詞進行比對
        tool_keywords = [keyword.strip() for keyword in tool_text.split(',')]
        job_keywords = [keyword.strip() for keyword in job_text.split(',')]

        # 查詢 DynamoDB 表格以獲取相關題目內容
        found_question_contents = []
        for keyword in tool_keywords + job_keywords:
            scan_kwargs = {
                'FilterExpression': Attr("tag").contains(keyword)
            }
            done = False
            start_key = None
            while not done:
                if start_key:
                    scan_kwargs['ExclusiveStartKey'] = start_key
                response = table.scan(**scan_kwargs)
                items = response.get('Items', [])
                if items:
                    for item in items:
                        question_content = item['題目內容']
                        found_question_contents.append(question_content)
                start_key = response.get('LastEvaluatedKey', None)
                done = start_key is None

        # 初始化人資和技術問題索引
        session['hr_index'] = session['tech_index'] = 0

        # 檢索人資問題
        scan_kwargs = {
            'FilterExpression': Attr("關鍵字").eq("人資")
        }
        hr_questions = []
        done = False
        start_key = None
        while not done:
            if start_key:
                scan_kwargs['ExclusiveStartKey'] = start_key
            response = table.scan(**scan_kwargs)
            hr_questions += response.get('Items', [])
            start_key = response.get('LastEvaluatedKey', None)
            done = start_key is None

        # 確保至少有三個人資問題，如果不足三個則全部使用
        session['hr_questions'] = random.sample(hr_questions, min(3, len(hr_questions))) if len(hr_questions) >= 3 else hr_questions

        # 確保至少有三個技術問題，如果不足三個則全部使用
        session['tech_questions'] = random.sample(found_question_contents, min(3, len(found_question_contents))) if found_question_contents else []

        # 重定向到第一個人資問題
        return redirect(url_for('hr_question'))

@app.route('/hr_question', methods=['GET', 'POST'])
def hr_question():
    # 從 session 中獲取人資問題和索引
    hr_questions = session.get('hr_questions', [])
    hr_index = session.get('hr_index', 0)
    
    # 檢查索引是否小於人資問題的數量
    if hr_index < len(hr_questions):
        # 獲取當前問題的內容
        question_content = hr_questions[hr_index]['題目內容']
        # 更新索引，準備下一個問題
        session['hr_index'] = hr_index + 1

        video_url = call_d_id(question_content)

        # 顯示人資問題的 HTML 模板
        return render_template('hr.html', question=question_content, video_url=video_url)
    else:
        # 如果所有人資問題都已回答完畢，將用戶重定向到技術問題
        return redirect(url_for('tech_question'))

@app.route('/tech_question', methods=['GET', 'POST'])
def tech_question():
    # 從 session 中獲取技術問題和索引
    tech_questions = session.get('tech_questions', [])
    tech_index = session.get('tech_index', 0)
    
    # 檢查索引是否小於技術問題的數量
    if tech_index < len(tech_questions):
        # 獲取當前問題的內容
        question = tech_questions[tech_index]
        # 更新索引，準備下一個問題
        session['tech_index'] = tech_index + 1
        # 顯示技術問題的 HTML 模板
        return render_template('tech.html', question=question)
    else:
        # 如果所有技術問題都已回答完畢，顯示信息，爾後是答題結束畫面
        return "所有問題都已答完，感謝您的參與！"
    
######################################
#             面試官影片             #
####################################
    
def call_d_id(question):
    url = 'https://api.d-id.com/talks'
    payload = {
        'script': {
            'type': 'text',
            'subtitles': 'false',
            'provider': {
                'type': 'microsoft',
                'voice_id': 'zh-TW-HsiaoChenNeural'
            },
            'ssml': 'false',
            'input': question
        },
        'config': {
            'fluent': 'false',
            'pad_audio': '0.0'
        },
        'source_url': 'https://create-images-results.d-id.com/DefaultPresenters/Emily_f/thumbnail.jpeg'
    }
    headers = {
        'accept': 'application/json',
        'content-type': 'application/json',
        'authorization': 'Basic YVhsMmIyNXVaVGt5TURVeE5rQm5iV0ZwYkM1amIyMDpDanBTN09lektndFVUWTNHNXJNZk4='
    }
    response = requests.post(url, json=payload, headers=headers)
    talk_id = response.json().get('id')
    delay = 3
    while True:
        url = f'https://api.d-id.com/talks/{talk_id}'
        headers = {
            'accept': 'application/json',
            'authorization': 'Basic YVhsMmIyNXVaVGt5TURVeE5rQm5iV0ZwYkM1amIyMDpDanBTN09lektndFVUWTNHNXJNZk4='
        }
        response = requests.get(url, headers=headers)
        response_data = response.json()
        if response_data.get('status') == 'done':
            result_url = response_data.get('result_url')
            break
        else:
            time.sleep(1)
    return result_url

if __name__ == '__main__':
    app.run(debug=True)