Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| import atproto as atp | |
| import markdown | |
| from langchain_cohere import ChatCohere | |
| from langchain_core.prompts import ChatPromptTemplate | |
| import os | |
| api_key = st.secrets["COHERE_API_KEY"] | |
| llm = ChatCohere(cohere_api_key=api_key) | |
| template = ChatPromptTemplate([ | |
| ( | |
| "system", | |
| "You are a helpful and efficient feed summarization assistant. You should provide a coincise and well-style, polished overview of what the post authors are telling" | |
| ), | |
| ( | |
| "human", | |
| "{message}" | |
| ) | |
| ]) | |
| chain = template | llm | |
| client = atp.Client() | |
| def text_inference(message): | |
| res = chain.invoke({"message": message}) | |
| ret = res.content | |
| return ret | |
| def get_feed(username: str, password: str): | |
| try: | |
| client.login(username, password) | |
| except Exception as e: | |
| return "### Your username or password are wrong :(" | |
| strtort = "" | |
| strsum = "" | |
| # Get "Home" page. Use pagination (cursor + limit) to fetch all posts | |
| timeline = client.get_timeline(algorithm='reverse-chronological') | |
| c = 0 | |
| for feed_view in timeline.feed: | |
| c+=1 | |
| action = 'New Post🆕' | |
| if feed_view.reason: | |
| action_by = feed_view.reason.by.handle | |
| action = f'Reposted by @{action_by}🔁' | |
| post = feed_view.post.record | |
| author = feed_view.post.author | |
| strtort += f'<h3>{action}</h3>\n<img alt="Avatar for {author.display_name}" src="{author.avatar}" width=50> <b>{author.display_name}<b>:<br><p>{post.text}</p><hr>' | |
| strsum += f'Author: {author.display_name} - Post content: {post.text}' | |
| if c>=25: | |
| break | |
| res = text_inference(strsum) | |
| reshtml = markdown.markdown(res) | |
| smry = f"\n\n<details>\n\t<summary><b>Feed Summary</b></summary>\n\n{reshtml}\n\n</details>\n\n" | |
| strtort = f"{smry}\n\n{strtort}" | |
| return strtort | |
| if __name__ == "__main__": | |
| # Title of the web app | |
| st.title("BlueSky User Feed🦋") | |
| st.subheader("Your home feed and its summary, in one place") | |
| st.image("https://upload.wikimedia.org/wikipedia/commons/thumb/7/7a/Bluesky_Logo.svg/1920px-Bluesky_Logo.svg.png", width=150) | |
| # Input text box for the search query | |
| username = st.text_input("You BlueSky username/handle", placeholder="user.bsky.social") | |
| password = st.text_input("Password", type="password") | |
| # Button to initiate search | |
| if st.button("Show my feed!"): | |
| if username != "" and password!="": | |
| results = get_feed(username, password) | |
| if results != "### Your username or password are wrong :(": | |
| st.write(f"## Home Feed (Following)🏠\n\n----------------------\n\n") | |
| st.html(results) | |
| else: | |
| st.write(results) | |
| else: | |
| if username == "" and password!="": | |
| st.write("### Please enter a username") | |
| elif password == "" and username!="": | |
| st.write("### Please enter a password") | |
| else: | |
| st.write("### Please enter a username and a password") | |