-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
108 lines (91 loc) · 3.57 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import JSONResponse
from starlette.middleware import Middleware
from time_middleware import TimeoutMiddleware
from typing import Optional
from pydantic import BaseModel
# Import the functions from the other scripts
from profile_data import retrieve_profile_info
from user_post import retrieve_post_metrics
from media import get_post_details
# Import the Instagram service
from instagram_service import instagram_service
from authentication_middleware import AuthenticationMiddleware
class Credentials(BaseModel):
username: str
password: str
app = FastAPI(middleware=[
Middleware(TimeoutMiddleware, timeout=600),
Middleware(AuthenticationMiddleware, instagram_service=instagram_service)
])
@app.post('/v1/api/login')
def login(credentials: Credentials):
try:
username = instagram_service.login(credentials.username, credentials.password)
return JSONResponse(content={"meassege": "Login Successful", "username": username})
except Exception as e:
raise HTTPException(
status_code=500,
detail=str(e)
)
@app.get('/v1/api/profile')
def profile_info(username: str = Query(..., description="Username of the profile to retrieve")):
try:
profile_data = retrieve_profile_info(username)
if profile_data is None:
raise HTTPException(status_code=404, detail="Profile not found")
return JSONResponse(content=profile_data)
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# Still needs work
@app.get('/v1/api/user_posts')
def user_posts(
username: str = Query(..., description="Username to retrieve posts for"),
from_date: Optional[str] = Query(None, description="Start date for post retrieval (YYYY-MM-DD), Optional"),
to_date: Optional[str] = Query(None, description="End date for post retrieval (YYYY-MM-DD), Optional")
):
try:
posts = retrieve_post_metrics(username, from_date, to_date)
if not posts:
raise HTTPException(status_code=404, detail="Posts for user not found")
return JSONResponse(content=posts)
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get('/v1/api/media')
def media_posts(url: str = Query(..., description="URL of the media to retrieve details for")):
try:
media_url = get_post_details(url)
if media_url is None:
raise HTTPException(status_code=404, detail="Media Info not found")
return JSONResponse(content=media_url)
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get('/v1/api/media_likers')
def media_likers(
media_id: str = Query(..., description="media_id = {post_id}_{owner_id} to retrieve likers for the post"),
username: str = Query(..., description="Username of the logged-in user")
):
try:
likers = instagram_service.get_media_likers(username, media_id)
if not likers:
raise HTTPException(
status_code=404,
detail="Likers for this media not found"
)
return JSONResponse(content=likers)
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(
status_code=500,
detail=str(e)
)
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000, log_level="debug")