kunthawat/ALwrity
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update ai_backlinking.py

Browse Source 
This script now includes the suggested improvements for:

Generating search queries
Finding backlink opportunities
Composing personalized emails
Sending emails using SMTP
Logging sent emails
Checking email responses
Sending follow-up emails
Handling multiple keywords
Main workflow integration
This commit is contained in:
ي
2025-01-17 09:44:03 +05:30
committed by GitHub
parent a339aa6b29
commit d92873ffdb
1 changed files with 172 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

189
lib/ai_marketing_tools/ai_backlinking.py
View File

@@ -132,6 +132,13 @@
import sys
from googlesearch import search
from loguru import logger
from lib.ai_web_researcher.firecrawl_web_crawler import scrape_website
from lib.gpt_providers.text_generation.main_text_generation import llm_text_gen
from lib.ai_web_researcher.firecrawl_web_crawler import scrape_url
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
# Configure logger
logger.remove()
logger.add(sys.stdout,
@@ -139,11 +146,6 @@ logger.add(sys.stdout,
format="<level>{level}</level>|<green>{file}:{line}:{function}</green>| {message}"
)
from lib.ai_web_researcher.firecrawl_web_crawler import scrape_website
from lib.gpt_providers.text_generation.main_text_generation import llm_text_gen
from lib.ai_web_researcher.firecrawl_web_crawler import scrape_url
def generate_search_queries(keyword):
"""
Generate a list of search queries for finding guest post opportunities.
@@ -156,16 +158,15 @@ def generate_search_queries(keyword):
"""
return [
f"{keyword} + 'Guest Contributor'",
# f"{keyword} + 'Add Guest Post'",
# f"{keyword} + 'Guest Bloggers Wanted'",
# f"{keyword} + 'Write for Us'",
# f"{keyword} + 'Submit Guest Post'",
# f"{keyword} + 'Become a Guest Blogger'",
# f"{keyword} + 'guest post opportunities'",
# f"{keyword} + 'Submit article'",
f"{keyword} + 'Add Guest Post'",
f"{keyword} + 'Guest Bloggers Wanted'",
f"{keyword} + 'Write for Us'",
f"{keyword} + 'Submit Guest Post'",
f"{keyword} + 'Become a Guest Blogger'",
f"{keyword} + 'guest post opportunities'",
f"{keyword} + 'Submit article'",
]
def find_backlink_opportunities(keyword):
"""
Find backlink opportunities by scraping websites based on search queries.
@@ -186,7 +187,7 @@ def find_backlink_opportunities(keyword):
logger.info(f"Scraped Website content for {url}: {website_data}")
if website_data:
contact_info = extract_contact_info(website_data)
logger.info("Contact details found for {url}: [contact_info]")
logger.info(f"Contact details found for {url}: {contact_info}")
# AI-driven insights using website data
insights_prompt = f"""
@@ -222,7 +223,6 @@ Website Content:
return results
def compose_personalized_email(website_data, insights, user_proposal):
"""
Compose a personalized outreach email using AI LLM based on website data, insights, and user proposal.
@@ -265,6 +265,40 @@ Please compose a professional and engaging email that includes:
return llm_text_gen(email_prompt)
def send_email(smtp_server, smtp_port, smtp_user, smtp_password, to_email, subject, body):
"""
Send an email using an SMTP server.
Args:
smtp_server (str): The SMTP server address.
smtp_port (int): The SMTP server port.
smtp_user (str): The SMTP server username.
smtp_password (str): The SMTP server password.
to_email (str): The recipient's email address.
subject (str): The email subject.
body (str): The email body.
Returns:
bool: True if the email was sent successfully, False otherwise.
"""
try:
msg = MIMEMultipart()
msg['From'] = smtp_user
msg['To'] = to_email
msg['Subject'] = subject
msg.attach(MIMEText(body, 'plain'))
server = smtplib.SMTP(smtp_server, smtp_port)
server.starttls()
server.login(smtp_user, smtp_password)
server.send_message(msg)
server.quit()
logger.info(f"Email sent successfully to {to_email}")
return True
except Exception as e:
logger.error(f"Failed to send email to {to_email}: {e}")
return False
def search_for_urls(query):
"""
@@ -282,8 +316,7 @@ def search_for_urls(query):
print(google_search_result)
return google_search_result
except Exception as err:
logger.error("Failed to do GoogleSearch: {err}")
logger.error(f"Failed to do GoogleSearch: {err}")
def extract_contact_info(website_data):
"""
@@ -300,3 +333,125 @@ def extract_contact_info(website_data):
"name": website_data.get("contact", {}).get("name", "Webmaster"),
"email": website_data.get("contact", {}).get("email", ""),
}
def find_backlink_opportunities_for_keywords(keywords):
"""
Find backlink opportunities for multiple keywords.
Args:
keywords (list): A list of keywords to search for backlink opportunities.
Returns:
dict: A dictionary with keywords as keys and a list of results as values.
"""
all_results = {}
for keyword in keywords:
results = find_backlink_opportunities(keyword)
all_results[keyword] = results
return all_results
def log_sent_email(keyword, email_info):
"""
Log the information of a sent email.
Args:
keyword (str): The keyword associated with the email.
email_info (dict): Information about the sent email (e.g., recipient, subject, body).
"""
with open(f"{keyword}_sent_emails.log", "a") as log_file:
log_file.write(f"{email_info}\n")
def check_email_responses(imap_server, imap_user, imap_password):
"""
Check email responses using an IMAP server.
Args:
imap_server (str): The IMAP server address.
imap_user (str): The IMAP server username.
imap_password (str): The IMAP server password.
Returns:
list: A list of email responses.
"""
responses = []
try:
mail = imaplib.IMAP4_SSL(imap_server)
mail.login(imap_user, imap_password)
mail.select('inbox')
status, data = mail.search(None, 'UNSEEN')
mail_ids = data[0]
id_list = mail_ids.split()
for mail_id in id_list:
status, data = mail.fetch(mail_id, '(RFC822)')
msg = email.message_from_bytes(data[0][1])
if msg.is_multipart():
for part in msg.walk():
if part.get_content_type() == 'text/plain':
responses.append(part.get_payload(decode=True).decode())
else:
responses.append(msg.get_payload(decode=True).decode())
mail.logout()
except Exception as e:
logger.error(f"Failed to check email responses: {e}")
return responses
def send_follow_up_email(smtp_server, smtp_port, smtp_user, smtp_password, to_email, subject, body):
"""
Send a follow-up email using an SMTP server.
Args:
smtp_server (str): The SMTP server address.
smtp_port (int): The SMTP server port.
smtp_user (str): The SMTP server username.
smtp_password (str): The SMTP server password.
to_email (str): The recipient's email address.
subject (str): The email subject.
body (str): The email body.
Returns:
bool: True if the email was sent successfully, False otherwise.
"""
return send_email(smtp_server, smtp_port, smtp_user, smtp_password, to_email, subject, body)
def main_backlinking_workflow(keywords, smtp_config, imap_config, user_proposal):
"""
Main workflow for the AI-powered backlinking feature.
Args:
keywords (list): A list of keywords to search for backlink opportunities.
smtp_config (dict): SMTP configuration for sending emails.
imap_config (dict): IMAP configuration for checking email responses.
user_proposal (dict): The user's proposal for a guest post or content contribution.
Returns:
None
"""
all_results = find_backlink_opportunities_for_keywords(keywords)
for keyword, results in all_results.items():
for result in results:
email_body = compose_personalized_email(result, result['insights'], user_proposal)
email_sent = send_email(
smtp_config['server'],
smtp_config['port'],
smtp_config['user'],
smtp_config['password'],
result['contact_info']['email'],
f"Guest Post Proposal for {result['metadata']['title']}",
email_body
)
if email_sent:
log_sent_email(keyword, {
"to": result['contact_info']['email'],
"subject": f"Guest Post Proposal for {result['metadata']['title']}",
"body": email_body
})
responses = check_email_responses(imap_config['server'], imap_config['user'], imap_config['password'])
for response in responses:
# TBD : Process and possibly send follow-up emails based on responses
pass