Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commit7bb752a

Browse files
committed
add detecting fraudulent transactions using stream app with kafka tutorial
1 parent272e14b commit7bb752a

File tree

9 files changed

+145
-0
lines changed

9 files changed

+145
-0
lines changed

‎README.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,7 @@ This is a repository of all the tutorials of [The Python Code](https://www.thepy
8686
-[How to Play and Record Audio in Python](https://www.thepythoncode.com/article/play-and-record-audio-sound-in-python). ([code](general/recording-and-playing-audio))
8787
-[How to Get Geographic Locations in Python](https://www.thepythoncode.com/article/get-geolocation-in-python). ([code](general/geolocation))
8888
-[How to Assembly, Disassembly and Emulate Machine Code using Python](https://www.thepythoncode.com/article/arm-x86-64-assembly-disassembly-and-emulation-in-python). ([code](general/assembly-code))
89+
-[Detecting Fraudulent Transactions in a Streaming Application using Kafka in Python](https://www.thepythoncode.com/article/detect-fraudulent-transactions-with-apache-kafka-in-python). ([code](general/detect-fraudulent-transactions))
8990

9091

9192
-###[Web Scraping](https://www.thepythoncode.com/topic/web-scraping)
Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
#[Detecting Fraudulent Transactions in a Streaming Application using Kafka in Python](https://www.thepythoncode.com/article/detect-fraudulent-transactions-with-apache-kafka-in-python)
2+
Check the original repo[here](https://github.com/bassemmarji/Kafka_Fraud_Detector).
Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
fromflaskimportFlask,Response,stream_with_context,render_template,json,url_for
2+
3+
fromkafkaimportKafkaConsumer
4+
fromsettingsimport*
5+
6+
# create the flask object app
7+
app=Flask(__name__)
8+
9+
defstream_template(template_name,**context):
10+
print('template name =',template_name)
11+
app.update_template_context(context)
12+
t=app.jinja_env.get_template(template_name)
13+
rv=t.stream(context)
14+
rv.enable_buffering(5)
15+
returnrv
16+
17+
defis_suspicious(transaction:dict)->bool:
18+
"""Determine whether a transaction is suspicious."""
19+
returntransaction["amount"]>=900
20+
21+
# this router will render the template named index.html and will pass the following parameters to it:
22+
# title and Kafka stream
23+
@app.route('/')
24+
defindex():
25+
defg():
26+
consumer=KafkaConsumer(
27+
TRANSACTIONS_TOPIC
28+
,bootstrap_servers=KAFKA_BROKER_URL
29+
,value_deserializer=lambdavalue:json.loads(value)
30+
,
31+
)
32+
formessageinconsumer:
33+
transaction:dict=message.value
34+
topic=FRAUD_TOPICifis_suspicious(transaction)elseLEGIT_TOPIC
35+
print(topic,transaction)# DEBUG
36+
yieldtopic,transaction
37+
38+
returnResponse(stream_template('index.html',title='Fraud Detector / Kafka',data=g()))
39+
40+
if__name__=="__main__":
41+
app.run(host="localhost" ,debug=True)
Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
importos
2+
importjson
3+
fromkafkaimportKafkaConsumer,KafkaProducer
4+
fromsettingsimport*
5+
6+
defis_suspicious(transaction:dict)->bool:
7+
"""Simple condition to determine whether a transaction is suspicious."""
8+
returntransaction["amount"]>=900
9+
10+
if__name__=="__main__":
11+
consumer=KafkaConsumer(
12+
TRANSACTIONS_TOPIC
13+
,bootstrap_servers=KAFKA_BROKER_URL
14+
,value_deserializer=lambdavalue:json.loads(value)
15+
,
16+
)
17+
18+
formessageinconsumer:
19+
transaction:dict=message.value
20+
topic=FRAUD_TOPICifis_suspicious(transaction)elseLEGIT_TOPIC
21+
print(topic,transaction)#DEBUG
Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
importos
2+
importjson
3+
fromtimeimportsleep
4+
fromkafkaimportKafkaProducer
5+
# import initialization parameters
6+
fromsettingsimport*
7+
fromtransactionsimportcreate_random_transaction
8+
9+
10+
if__name__=="__main__":
11+
producer=KafkaProducer(bootstrap_servers=KAFKA_BROKER_URL
12+
#Encode all values as JSON
13+
,value_serializer=lambdavalue:json.dumps(value).encode()
14+
,)
15+
whileTrue:
16+
transaction:dict=create_random_transaction()
17+
producer.send(TRANSACTIONS_TOPIC,value=transaction)
18+
print(transaction)#DEBUG
19+
sleep(SLEEP_TIME)
Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
Kafka-python==2.0.2
2+
Flask==1.1.2
Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
# URL for our broker used for connecting to the Kafka cluster
2+
KAFKA_BROKER_URL="localhost:9092"
3+
# name of the topic hosting the transactions to be processed and requiring processing
4+
TRANSACTIONS_TOPIC="queuing.transactions"
5+
# these 2 variables will control the amount of transactions automatically generated
6+
TRANSACTIONS_PER_SECOND=float("2.0")
7+
SLEEP_TIME=1/TRANSACTIONS_PER_SECOND
8+
# name of the topic hosting the legitimate transactions
9+
LEGIT_TOPIC="queuing.legit"
10+
# name of the topic hosting the suspicious transactions
11+
FRAUD_TOPIC="queuing.fraud"
Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
<!doctype html>
2+
<title> Send Javascript with template demo</title>
3+
<html>
4+
5+
<head>
6+
</head>
7+
8+
<body>
9+
<divclass="container">
10+
<h1>{{title}}</h1>
11+
</div>
12+
<divid="data"></div>
13+
{% for topic, transaction in data: %}
14+
<script>
15+
vartopic="{{ topic }}";
16+
vartransaction="{{ transaction }}";
17+
if(topic.search("fraud")>0){
18+
topic=topic.fontcolor("red")
19+
}else{
20+
topic=topic.fontcolor("green")
21+
}
22+
document.getElementById('data').innerHTML+="<br>"+topic+" "+transaction;
23+
</script>
24+
{% endfor %}
25+
</body>
26+
27+
</html>
Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
fromrandomimportchoices,randint
2+
fromstringimportascii_letters,digits
3+
4+
account_chars:str=digits+ascii_letters
5+
6+
def_random_account_id()->str:
7+
"""Return a random account number made of 12 characters"""
8+
return"".join(choices(account_chars,k=12))
9+
10+
def_random_amount()->float:
11+
"""Return a random amount between 1.00 and 1000.00"""
12+
returnrandint(100,1000000)/100
13+
14+
defcreate_random_transaction()->dict:
15+
"""Create a fake randomised transaction."""
16+
return {
17+
"source":_random_account_id()
18+
,"target":_random_account_id()
19+
,"amount":_random_amount()
20+
,"currency":"EUR"
21+
}

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp